Fix GCSToGCSOperator ignoring replace=False for single-file copy

providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_gcs.py

@@ -416,11 +416,20 @@ def _copy_source_without_wildcard(self, hook, prefix) -> list[str]:
         # and copy directly
         if len(objects) == 0 and prefix:
             if hook.exists(self.source_bucket, prefix):
-                uri = self._copy_single_object(
-                    hook=hook, source_object=prefix, destination_object=self.destination_object
-                )
-                if uri:
-                    result_uris.append(uri)
+                # `objects` may have been emptied by _ignore_existing_files; respect replace=False here too.
+                destination_object = self.destination_object or prefix
+                if not self.replace and hook.exists(self.destination_bucket, destination_object):
+                    self.log.info(
+                        "Skipped object %s; already exists in destination bucket %s.",
+                        destination_object,
+                        self.destination_bucket,
+                    )
+                else:
+                    uri = self._copy_single_object(
+                        hook=hook, source_object=prefix, destination_object=self.destination_object
+                    )
+                    if uri:
+                        result_uris.append(uri)
             elif self.source_object_required:
                 msg = f"{prefix} does not exist in bucket {self.source_bucket}"
                 self.log.warning(msg)

providers/google/tests/unit/google/cloud/transfers/test_gcs_to_gcs.py

@@ -677,6 +677,68 @@ def test_execute_source_object_required_flag_true(self, mock_hook):
         ):
             operator.execute(None)
 
+    @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+    def test_execute_replace_false_skips_single_file_when_destination_exists(self, mock_hook):
+        mock_hook.return_value.list.side_effect = [
+            [SOURCE_OBJECT_NO_WILDCARD],
+            [SOURCE_OBJECT_NO_WILDCARD],
+        ]
+        mock_hook.return_value.exists.return_value = True
+
+        operator = GCSToGCSOperator(
+            task_id=TASK_ID,
+            source_bucket=TEST_BUCKET,
+            source_object=SOURCE_OBJECT_NO_WILDCARD,
+            destination_bucket=DESTINATION_BUCKET,
+            destination_object=SOURCE_OBJECT_NO_WILDCARD,
+            replace=False,
+        )
+
+        operator.execute(None)
+
+        mock_hook.return_value.exists.assert_any_call(DESTINATION_BUCKET, SOURCE_OBJECT_NO_WILDCARD)
+        mock_hook.return_value.rewrite.assert_not_called()
+
+    @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+    def test_execute_replace_false_copies_single_file_when_destination_missing(self, mock_hook):
+        mock_hook.return_value.list.side_effect = [[], []]
+        mock_hook.return_value.exists.side_effect = lambda bucket, _obj: bucket == TEST_BUCKET
+
+        operator = GCSToGCSOperator(
+            task_id=TASK_ID,
+            source_bucket=TEST_BUCKET,
+            source_object=SOURCE_OBJECT_NO_WILDCARD,
+            destination_bucket=DESTINATION_BUCKET,
+            destination_object=SOURCE_OBJECT_NO_WILDCARD,
+            replace=False,
+        )
+
+        operator.execute(None)
+
+        mock_hook.return_value.rewrite.assert_called_once_with(
+            TEST_BUCKET, SOURCE_OBJECT_NO_WILDCARD, DESTINATION_BUCKET, SOURCE_OBJECT_NO_WILDCARD
+        )
+
+    @mock.patch("airflow.providers.google.cloud.transfers.gcs_to_gcs.GCSHook")
+    def test_execute_replace_true_copies_single_file_even_when_destination_exists(self, mock_hook):
+        mock_hook.return_value.list.return_value = []
+        mock_hook.return_value.exists.return_value = True
+
+        operator = GCSToGCSOperator(
+            task_id=TASK_ID,
+            source_bucket=TEST_BUCKET,
+            source_object=SOURCE_OBJECT_NO_WILDCARD,
+            destination_bucket=DESTINATION_BUCKET,
+            destination_object=SOURCE_OBJECT_NO_WILDCARD,
+            replace=True,
+        )
+
+        operator.execute(None)
+
+        mock_hook.return_value.rewrite.assert_called_once_with(
+            TEST_BUCKET, SOURCE_OBJECT_NO_WILDCARD, DESTINATION_BUCKET, SOURCE_OBJECT_NO_WILDCARD
+        )
+
     @pytest.mark.parametrize(
         (
             "existing_objects",