Fix running tests with dbt Fusion 2.0.0 preview versions

.github/workflows/test.yml

@@ -369,6 +369,8 @@ jobs:
           POSTGRES_PORT: 5432
           AIRFLOW__COSMOS__REMOTE_TARGET_PATH: "s3://cosmos-remote-cache/target_compiled/"
           AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID: aws_s3_conn
+          DBT_PROJECT_NAME: "altered_jaffle_shop"  # The syntax of jaffle_shop is not supported in dbt 1.5.4, only in dbt >= 1.10
+          TEST_SINGLE_DAG: "basic_cosmos_task_group.py"
 
   Run-Integration-Tests-DBT-Async:
     needs: Authorize
@@ -380,7 +382,7 @@ jobs:
       matrix:
         python-version: [ "3.11" ]
         airflow-version: [ "2.11", "3.0" ]
-        dbt-version: ["1.5", "1.6", "1.7", "1.8", "1.9", "1.10"]
+        dbt-version: ["1.6", "1.7", "1.8", "1.9", "1.10"]
     services:
       postgres:
         image: postgres@sha256:4cd697181d4bd3ddc41a09012f339fa8cb5a8cd3d8b30130ea8378c176b6c494  # 14.18
@@ -421,16 +423,6 @@ jobs:
       - name: Test Cosmos against Airflow ${{ matrix.airflow-version }}, Python ${{ matrix.python-version }} and dbt ${{ matrix.dbt-version }}
         run: |
           hatch run tests.py${{ matrix.python-version }}-${{ matrix.airflow-version }}-${{ matrix.dbt-version }}:test-integration-dbt-async
-        env:
-          AIRFLOW_HOME: /home/runner/work/astronomer-cosmos/astronomer-cosmos/
-          AIRFLOW_CONN_AWS_S3_CONN: ${{ secrets.AIRFLOW_CONN_AWS_S3_CONN }}
-          AIRFLOW_CONN_GCP_GS_CONN: ${{ secrets.AIRFLOW_CONN_GCP_GS_CONN }}
-          AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 90.0
-          PYTHONPATH: /home/runner/work/astronomer-cosmos/astronomer-cosmos/:$PYTHONPATH
-          AIRFLOW__COSMOS__ENABLE_CACHE: 0
-          AIRFLOW__COSMOS__REMOTE_TARGET_PATH: "s3://cosmos-remote-cache/target_compiled/"
-          AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID: aws_s3_conn
-          DBT_ADAPTER_VERSION: ${{ matrix.dbt-version }}
 
       - name: Upload coverage to Github
         uses: actions/upload-artifact@v4
@@ -441,9 +433,27 @@ jobs:
 
     env:
       AIRFLOW_HOME: /home/runner/work/astronomer-cosmos/astronomer-cosmos/
-      AIRFLOW_CONN_EXAMPLE_CONN: postgres://postgres:postgres@0.0.0.0:5432/postgres
       PYTHONPATH: /home/runner/work/astronomer-cosmos/astronomer-cosmos/:$PYTHONPATH
 
+      DBT_PROJECT_NAME: "altered_jaffle_shop" # The syntax of jaffle_shop is not supported in dbt 1.5.4, only in dbt >= 1.10
+      AIRFLOW__COSMOS__ENABLE_DATASET_ALIAS: 0
+      AIRFLOW_CONN_AWS_S3_CONN: ${{ secrets.AIRFLOW_CONN_AWS_S3_CONN }}
+      AIRFLOW_CONN_GCP_GS_CONN: ${{ secrets.AIRFLOW_CONN_GCP_GS_CONN }}
+      AIRFLOW__CORE__DAGBAG_IMPORT_TIMEOUT: 90.0
+      AIRFLOW__COSMOS__ENABLE_CACHE: 0
+      AIRFLOW__COSMOS__REMOTE_TARGET_PATH: "s3://cosmos-remote-cache/target_compiled/"
+      AIRFLOW__COSMOS__REMOTE_TARGET_PATH_CONN_ID: aws_s3_conn
+      DBT_ADAPTER_VERSION: ${{ matrix.dbt-version }}
+
+      # The following are only required because the profiles.yml file references them. They are not used by the tests selected by this job.
+      AIRFLOW_CONN_EXAMPLE_CONN: postgres://postgres:postgres@0.0.0.0:5432/postgres
+      POSTGRES_HOST: localhost
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+      POSTGRES_DB: postgres
+      POSTGRES_SCHEMA: public
+      POSTGRES_PORT: 5432
+
   Run-Integration-dbt-fusion-Tests:
     needs: Authorize
     runs-on: ubuntu-latest
@@ -614,7 +624,7 @@ jobs:
       matrix:
         python-version: [ "3.12" ]
         airflow-version: [ "2.10", "3.0" ]
-        dbt-version: [ "1.9" ]
+        dbt-version: [ "1.10" ]
     steps:
       - uses: actions/checkout@v5
         with:

dev/dags/basic_cosmos_dag.py

@@ -14,6 +14,9 @@
 
 DEFAULT_DBT_ROOT_PATH = Path(__file__).parent / "dbt"
 DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
+DBT_PROJECT_NAME = os.getenv("DBT_PROJECT_NAME", "jaffle_shop")
+DBT_PROJECT_PATH = DBT_ROOT_PATH / DBT_PROJECT_NAME
+
 
 profile_config = ProfileConfig(
     profile_name="default",
@@ -28,9 +31,7 @@
 # [START local_example]
 basic_cosmos_dag = DbtDag(
     # dbt/cosmos-specific parameters
-    project_config=ProjectConfig(
-        DBT_ROOT_PATH / "jaffle_shop",
-    ),
+    project_config=ProjectConfig(DBT_PROJECT_PATH),
     profile_config=profile_config,
     operator_args={
         "install_deps": True,  # install any necessary dependencies before running any dbt command

dev/dags/basic_cosmos_dag_full_module_path_imports.py

@@ -15,6 +15,8 @@
 
 DEFAULT_DBT_ROOT_PATH = Path(__file__).parent / "dbt"
 DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
+DBT_PROJECT_NAME = os.getenv("DBT_PROJECT_NAME", "jaffle_shop")
+DBT_PROJECT_PATH = DBT_ROOT_PATH / DBT_PROJECT_NAME
 
 profile_config = ProfileConfig(
     profile_name="default",
@@ -30,7 +32,7 @@
 basic_cosmos_dag_full_module_path_imports = DbtDag(
     # dbt/cosmos-specific parameters
     project_config=ProjectConfig(
-        DBT_ROOT_PATH / "jaffle_shop",
+        DBT_PROJECT_PATH,
     ),
     profile_config=profile_config,
     operator_args={

dev/dags/basic_cosmos_task_group.py

@@ -15,6 +15,8 @@
 
 DEFAULT_DBT_ROOT_PATH = Path(__file__).resolve().parent / "dbt"
 DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
+DBT_PROJECT_NAME = os.getenv("DBT_PROJECT_NAME", "jaffle_shop")
+DBT_PROJECT_PATH = DBT_ROOT_PATH / DBT_PROJECT_NAME
 
 profile_config = ProfileConfig(
     profile_name="default",
@@ -43,7 +45,7 @@
 
     customers = DbtTaskGroup(
         group_id="customers",
-        project_config=ProjectConfig((DBT_ROOT_PATH / "jaffle_shop").as_posix(), dbt_vars={"var": "2"}),
+        project_config=ProjectConfig((DBT_PROJECT_PATH).as_posix(), dbt_vars={"var": "2"}),
         render_config=RenderConfig(
             select=["path:seeds/raw_customers.csv"],
             enable_mock_profile=False,
@@ -59,7 +61,7 @@
     orders = DbtTaskGroup(
         group_id="orders",
         project_config=ProjectConfig(
-            (DBT_ROOT_PATH / "jaffle_shop").as_posix(),
+            (DBT_PROJECT_PATH).as_posix(),
         ),
         render_config=RenderConfig(
             select=["path:seeds/raw_orders.csv"],

dev/dags/dbt/jaffle_shop/dbt_project.yml

@@ -21,6 +21,6 @@ require-dbt-version: [">=1.0.0", "<2.0.0"]
 
 models:
   jaffle_shop:
-      materialized: table
+      +materialized: table
       staging:
-        materialized: view
+        +materialized: view

dev/dags/dbt/jaffle_shop/models/schema.yml

@@ -55,7 +55,8 @@ models:
         description: '{{ doc("orders_status") }}'
         tests:
           - accepted_values:
-              values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']
+              arguments:
+                values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']
 
       - name: amount
         description: Total amount (AUD) of the order

dev/dags/dbt/jaffle_shop/models/staging/schema.yml

@@ -17,7 +17,8 @@ models:
       - name: status
         tests:
           - accepted_values:
-              values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']
+              arguments:
+                values: ['placed', 'shipped', 'completed', 'return_pending', 'returned']
 
   - name: stg_payments
     columns:
@@ -28,4 +29,5 @@ models:
       - name: payment_method
         tests:
           - accepted_values:
-              values: ['credit_card', 'coupon', 'bank_transfer', 'gift_card']
+              arguments:
+                values: ['credit_card', 'coupon', 'bank_transfer', 'gift_card']

dev/dags/simple_dag_async.py

@@ -8,6 +8,8 @@
 
 DEFAULT_DBT_ROOT_PATH = Path(__file__).resolve().parent / "dbt"
 DBT_ROOT_PATH = Path(os.getenv("DBT_ROOT_PATH", DEFAULT_DBT_ROOT_PATH))
+DBT_PROJECT_NAME = os.getenv("DBT_PROJECT_NAME", "jaffle_shop")
+DBT_PROJECT_PATH = DBT_ROOT_PATH / DBT_PROJECT_NAME
 
 DBT_ADAPTER_VERSION = os.getenv("DBT_ADAPTER_VERSION", "1.9")
 
@@ -24,7 +26,7 @@
 simple_dag_async = DbtDag(
     # dbt/cosmos-specific parameters
     project_config=ProjectConfig(
-        DBT_ROOT_PATH / "jaffle_shop",
+        DBT_PROJECT_PATH,
     ),
     profile_config=profile_config,
     execution_config=ExecutionConfig(

scripts/test/integration-dbt-1-5-4.sh

@@ -27,4 +27,4 @@ pytest -vv \
     -m 'integration and not dbtFusion' \
     --ignore=tests/perf \
     --ignore=tests/test_example_k8s_dags.py \
-    -k 'basic_cosmos_task_group'
+    "tests/test_example_dags.py::test_example_dag[basic_cosmos_task_group]"

scripts/test/integration.sh

@@ -22,4 +22,4 @@ pytest -vv \
     --ignore=tests/perf \
     --ignore=tests/test_async_example_dag.py \
     --ignore=tests/test_example_k8s_dags.py \
-    -k 'not ( example_cosmos_python_models or example_virtualenv or jaffle_shop_kubernetes)'
+    -k 'not (simple_dag_async or example_cosmos_python_models or example_virtualenv or jaffle_shop_kubernetes)'

tests/dbt/test_graph.py

@@ -1908,9 +1908,9 @@ def test_save_dbt_ls_cache(mock_variable_set, mock_datetime, tmp_dbt_project_dir
     assert hash_args == "d41d8cd98f00b204e9800998ecf8427e"
     if sys.platform == "darwin":
         # We faced inconsistent hashing versions depending on the version of MacOS/Linux - the following line aims to address these.
-        assert hash_dir in ("e16e6aef5f03f4e12d744b9412d90272", "71bbf303ad4e06a7b1e2be20e0b73c0d")
+        assert hash_dir in ("481324dabe926f5cf6352b05e5ebe5d7", "60c08a4730a39d03d89f0f87a8ff3931")
     else:
-        assert hash_dir == "71bbf303ad4e06a7b1e2be20e0b73c0d"
+        assert hash_dir == "60c08a4730a39d03d89f0f87a8ff3931"
 
 
 @pytest.mark.integration

tests/test_example_dags.py

@@ -72,7 +72,6 @@ def get_dag_bag() -> DagBag:  # noqa: C901
             file.writelines([f"{dagfile}\n"])
 
         if DBT_VERSION < Version("1.6.0"):
-            file.writelines(["simple_dag_async.py\n"])
             file.writelines(["example_model_version.py\n"])
             file.writelines(["example_operators.py\n"])
 
@@ -97,13 +96,40 @@ def get_dag_bag() -> DagBag:  # noqa: C901
     return db
 
 
+def get_dag_bag_single_dag(single_dag: str) -> DagBag:
+    """Create a DagBag by adding the files that are not supported to .airflowignore"""
+    # add everything to airflow ignore that isn't performance_dag.py
+    with open(AIRFLOW_IGNORE_FILE, "w+") as f:
+        for file in EXAMPLE_DAGS_DIR.iterdir():
+            if file.is_file() and file.suffix == ".py":
+                if file.name != single_dag:
+                    print(f"Adding {file.name} to .airflowignore")
+                    f.write(f"{file.name}\n")
+    print(".airflowignore contents: ")
+    print(AIRFLOW_IGNORE_FILE.read_text())
+    db = DagBag(EXAMPLE_DAGS_DIR, include_examples=False)
+    assert db.dags
+    assert not db.import_errors
+    return db
+
+
+def get_dagbag_depending_on_single_dag() -> DagBag:
+    """Return DagBag for a single DAG or for all DAGs, depending on environment variable TEST_SINGLE_DAG"""
+    single_dag = os.getenv("TEST_SINGLE_DAG")
+    if single_dag:
+        dag_bag = get_dag_bag_single_dag(single_dag)
+    else:
+        dag_bag = get_dag_bag()
+    return dag_bag
+
+
 def get_dag_ids() -> list[str]:
-    dag_bag = get_dag_bag()
+    dag_bag = get_dagbag_depending_on_single_dag()
     return dag_bag.dag_ids
 
 
 def run_dag(dag_id: str):
-    dag_bag = get_dag_bag()
+    dag_bag = get_dagbag_depending_on_single_dag()
     dag = dag_bag.get_dag(dag_id)
     assert dag
     test_utils.run_dag(dag)