Fixed CloudComposerDAGRunSensor to return False when no runs exist in execution_range

providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py

@@ -231,16 +231,17 @@ def _check_dag_runs_states(
         start_date: datetime,
         end_date: datetime,
     ) -> bool:
+        found_runs_in_window = False
         for dag_run in dag_runs:
-            if (
-                start_date.timestamp()
-                < parser.parse(
-                    dag_run["execution_date" if self._composer_airflow_version < 3 else "logical_date"]
-                ).timestamp()
-                < end_date.timestamp()
-            ) and dag_run["state"] not in self.allowed_states:
-                return False
-        return True
+            execution_date = parser.parse(
+                dag_run["execution_date" if self._composer_airflow_version < 3 else "logical_date"]
+            )
+
+            if start_date.timestamp() < execution_date.timestamp() < end_date.timestamp():
+                found_runs_in_window = True
+                if dag_run["state"] not in self.allowed_states:
+                    return False
+        return found_runs_in_window
 
     def _get_composer_airflow_version(self) -> int:
         """Return Composer Airflow version."""

providers/google/tests/unit/google/cloud/sensors/test_cloud_composer.py

@@ -205,6 +205,33 @@ def test_composer_dag_run_id_wait_not_ready(
 
         assert not task.poke(context={"logical_date": datetime(2024, 5, 23, 0, 0, 0)})
 
+    @pytest.mark.parametrize("use_rest_api", [True, False])
+    @pytest.mark.parametrize("composer_airflow_version", [2, 3])
+    @mock.patch("airflow.providers.google.cloud.sensors.cloud_composer.ExecuteAirflowCommandResponse.to_dict")
+    @mock.patch("airflow.providers.google.cloud.sensors.cloud_composer.CloudComposerHook")
+    def test_dag_run_outside_execution_range(
+        self, mock_hook, to_dict_mode, composer_airflow_version, use_rest_api
+    ):
+        mock_hook.return_value.wait_command_execution_result.return_value = TEST_EXEC_RESULT(
+            "success", "execution_date" if composer_airflow_version < 3 else "logical_date"
+        )
+        mock_hook.return_value.get_dag_runs.return_value = TEST_GET_RESULT(
+            "success", "execution_date" if composer_airflow_version < 3 else "logical_date"
+        )
+        task = CloudComposerDAGRunSensor(
+            task_id="task-id",
+            project_id=TEST_PROJECT_ID,
+            region=TEST_REGION,
+            environment_id=TEST_ENVIRONMENT_ID,
+            composer_dag_id="test_dag_id",
+            execution_range=[datetime(2024, 5, 22, 17, 0, 0), datetime(2024, 5, 22, 20, 0, 0)],
+            allowed_states=["success"],
+            use_rest_api=use_rest_api,
+        )
+        task._composer_airflow_version = composer_airflow_version
+
+        assert not task.poke(context={"logical_date": datetime(2024, 5, 23, 0, 0, 0)})
+
 
 class TestCloudComposerExternalTaskSensor:
     @pytest.mark.parametrize("composer_airflow_version", [2, 3])