diff --git a/airflow/providers/dbt/cloud/operators/dbt.py b/airflow/providers/dbt/cloud/operators/dbt.py
index 0b31ba30145fd..8e9f310ce8e5b 100644
--- 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
index 2853a73bae6d6..b1a667aea1691 100644
--- 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(