fix(dbt): fix wrong payload set when reuse_existing_run set to True i…

apache · Apr 26, 2024 · d1fe16f · d1fe16f
1 parent 8188da7
commit d1fe16f
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 8 deletions.
diff --git a/airflow/providers/dbt/cloud/operators/dbt.py b/airflow/providers/dbt/cloud/operators/dbt.py
@@ -135,7 +135,7 @@ def execute(self, context: Context):
                 account_id=self.account_id,
                 payload={
                     "job_definition_id": self.job_id,
-                    "status": DbtCloudJobRunStatus.NON_TERMINAL_STATUSES,
+                    "status__in": DbtCloudJobRunStatus.NON_TERMINAL_STATUSES,
                     "order_by": "-created_at",
                 },
             ).json()["data"]

diff --git a/tests/providers/dbt/cloud/operators/test_dbt.py b/tests/providers/dbt/cloud/operators/test_dbt.py
@@ -312,9 +312,9 @@ def test_execute_no_wait_for_termination(self, mock_run_job, conn_id, account_id
         ids=["default_account", "explicit_account"],
     )
     def test_execute_no_wait_for_termination_and_reuse_existing_run(
-        self, mock_run_job, mock_get_jobs_run, conn_id, account_id
+        self, mock_run_job, mock_get_job_runs, conn_id, account_id
     ):
-        mock_get_jobs_run.return_value.json.return_value = {
+        mock_get_job_runs.return_value.json.return_value = {
             "data": [
                 {
                     "id": 10000,
@@ -354,12 +354,17 @@ def test_execute_no_wait_for_termination_and_reuse_existing_run(
         assert operator.schema_override == self.config["schema_override"]
         assert operator.additional_run_config == self.config["additional_run_config"]
 
-        with patch.object(DbtCloudHook, "get_job_run") as mock_get_job_run:
-            operator.execute(context=self.mock_context)
-
-            mock_run_job.assert_not_called()
+        operator.execute(context=self.mock_context)
 
-            mock_get_job_run.assert_not_called()
+        mock_run_job.assert_not_called()
+        mock_get_job_runs.assert_called_with(
+            account_id=account_id,
+            payload={
+                "job_definition_id": self.config["job_id"],
+                "status__in": DbtCloudJobRunStatus.NON_TERMINAL_STATUSES,
+                "order_by": "-created_at",
+            },
+        )
 
     @patch.object(DbtCloudHook, "trigger_job_run")
     @pytest.mark.parametrize(