Improve tests for callback / dbt project fixture

cosmos/operators/local.py

@@ -275,7 +275,7 @@ def calculate_openlineage_events_completes(
         try:
             events = openlineage_processor.parse()
             self.openlineage_events_completes = events.completes
-        except (FileNotFoundError, NotImplementedError):
+        except (FileNotFoundError, NotImplementedError, ValueError):
             logger.debug("Unable to parse OpenLineage events", stack_info=True)
 
     def get_datasets(self, source: Literal["inputs", "outputs"]) -> list[Dataset]:

tests/operators/test_local.py

@@ -31,8 +31,9 @@
 
 
 DBT_PROJ_DIR = Path(__file__).parent.parent.parent / "dev/dags/dbt/jaffle_shop"
-SCHEMA_FAILING_TEST = Path(__file__).parent.parent / "sample/schema_failing_test.yml"
-
+MINI_DBT_PROJ_DIR = Path(__file__).parent.parent / "sample/mini"
+MINI_DBT_PROJ_DIR_FAILING_SCHEMA = MINI_DBT_PROJ_DIR / "schema_failing_test.yml"
+MINI_DBT_PROJ_PROFILE = MINI_DBT_PROJ_DIR / "profiles.yml"
 
 profile_config = ProfileConfig(
     profile_name="default",
@@ -49,15 +50,17 @@
     ),
 )
 
+mini_profile_config = ProfileConfig(profile_name="mini", target_name="dev", profiles_yml_filepath=MINI_DBT_PROJ_PROFILE)
+
 
 @pytest.fixture
 def failing_test_dbt_project(tmp_path):
     tmp_dir = tempfile.TemporaryDirectory()
-    tmp_dir_path = Path(tmp_dir.name) / "jaffle_shop"
-    shutil.copytree(DBT_PROJ_DIR, tmp_dir_path)
+    tmp_dir_path = Path(tmp_dir.name) / "mini"
+    shutil.copytree(MINI_DBT_PROJ_DIR, tmp_dir_path)
     target_schema = tmp_dir_path / "models/schema.yml"
-    os.remove(target_schema)
-    shutil.copy(SCHEMA_FAILING_TEST, target_schema)
+    target_schema.exists() and os.remove(target_schema)
+    shutil.copy(MINI_DBT_PROJ_DIR_FAILING_SCHEMA, target_schema)
     yield tmp_dir_path
     tmp_dir.cleanup()
 
@@ -179,14 +182,14 @@ def test_run_operator_dataset_inlets_and_outlets():
 
     with DAG("test-id-1", start_date=datetime(2022, 1, 1)) as dag:
         run_operator = DbtRunLocalOperator(
-            profile_config=real_profile_config,
+            profile_config=mini_profile_config,
             project_dir=DBT_PROJ_DIR,
             task_id="run",
             dbt_cmd_flags=["--models", "stg_customers"],
             install_deps=True,
         )
         test_operator = DbtTestLocalOperator(
-            profile_config=real_profile_config,
+            profile_config=mini_profile_config,
             project_dir=DBT_PROJ_DIR,
             task_id="test",
             dbt_cmd_flags=["--models", "stg_customers"],
@@ -205,19 +208,17 @@ def test_run_test_operator_with_callback(failing_test_dbt_project):
     on_warning_callback = MagicMock()
 
     with DAG("test-id-2", start_date=datetime(2022, 1, 1)) as dag:
-        run_operator = DbtRunLocalOperator(
-            profile_config=real_profile_config,
+        run_operator = DbtSeedLocalOperator(
+            profile_config=mini_profile_config,
             project_dir=failing_test_dbt_project,
             task_id="run",
-            dbt_cmd_flags=["--models", "orders"],
-            install_deps=True,
+            append_env=True,
         )
         test_operator = DbtTestLocalOperator(
-            profile_config=real_profile_config,
+            profile_config=mini_profile_config,
             project_dir=failing_test_dbt_project,
             task_id="test",
-            dbt_cmd_flags=["--models", "orders"],
-            install_deps=True,
+            append_env=True,
             on_warning_callback=on_warning_callback,
         )
         run_operator >> test_operator
@@ -230,20 +231,18 @@ def test_run_test_operator_without_callback():
     on_warning_callback = MagicMock()
 
     with DAG("test-id-3", start_date=datetime(2022, 1, 1)) as dag:
-        run_operator = DbtRunLocalOperator(
-            profile_config=real_profile_config,
-            project_dir=DBT_PROJ_DIR,
+        run_operator = DbtSeedLocalOperator(
+            profile_config=mini_profile_config,
+            project_dir=MINI_DBT_PROJ_DIR,
             task_id="run",
-            dbt_cmd_flags=["--models", "orders"],
-            install_deps=True,
+            append_env=True,
         )
         test_operator = DbtTestLocalOperator(
-            profile_config=real_profile_config,
-            project_dir=DBT_PROJ_DIR,
+            profile_config=mini_profile_config,
+            project_dir=MINI_DBT_PROJ_DIR,
             task_id="test",
-            dbt_cmd_flags=["--models", "orders"],
-            install_deps=True,
-            exclude="relationships_orders_customer_id__customer_id__ref_customers_",
+            append_env=True,
+            on_warning_callback=on_warning_callback,
         )
         run_operator >> test_operator
     run_test_dag(dag)

tests/sample/mini/dbt_project.yml

@@ -0,0 +1,20 @@
+name: 'mini'
+
+config-version: 2
+version: '0.1'
+
+profile: 'mini'
+
+model-paths: ["models"]
+seed-paths: ["seeds"]
+test-paths: ["tests"]
+analysis-paths: ["analysis"]
+macro-paths: ["macros"]
+
+target-path: "target"
+clean-targets:
+    - "target"
+    - "dbt_modules"
+    - "logs"
+
+require-dbt-version: [">=1.0.0", "<2.0.0"]

tests/sample/mini/profiles.yml

@@ -0,0 +1,12 @@
+mini:
+  target: dev
+  outputs:
+    dev:
+      type: postgres
+      host: "{{ env_var('POSTGRES_HOST') }}"
+      user: "{{ env_var('POSTGRES_USER') }}"
+      password: "{{ env_var('POSTGRES_PASSWORD') }}"
+      port: "{{ env_var('POSTGRES_PORT') | int }}"
+      dbname: "{{ env_var('POSTGRES_DB') }}"
+      schema: "{{ env_var('POSTGRES_SCHEMA') }}"
+      threads: 4

tests/sample/schema_failing_test.yml → tests/sample/mini/schema_failing_test.yml

@@ -1,17 +1,17 @@
 version: 2
 
-models:
+seeds:
 
-  - name: orders
+  - name: mini_orders
     description: This table has basic information about orders, as well as some derived facts based on payments
 
     columns:
 
       - name: status
-        description: '{{ doc("orders_status") }}'
+        description: 'Order status'
         tests:
           - accepted_values:
-              # this test will fail, since this column has more values
+              # this will intentionally fail, since the seed has other values for this column
               values: ['placed']
               config:
                 severity: warn

tests/sample/mini/seeds/mini_orders.csv

@@ -0,0 +1,10 @@
+id,user_id,order_date,status
+1,1,2018-01-01,returned
+2,3,2018-01-02,completed
+3,22,2018-01-26,return_pending
+4,9,2018-03-17,shipped
+75,69,2018-03-18,completed
+76,25,2018-03-20,completed
+77,35,2018-03-21,shipped
+78,90,2018-03-23,shipped
+6,68,2018-03-26,placed