Fix silent overwrite of colliding blobs when replace=True

Author: yuseok89

providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py

@@ -188,33 +188,37 @@ def execute(self, context: Context) -> list[str]:
         if not self.keep_directory_structure and self.prefix and not self.flatten_structure:
             blob_prefix = os.path.join(blob_prefix, self.prefix)
 
+        existing_blobs_set: set[str] = set()
         if not self.replace:
-            existing_blobs = wasb_hook.get_blobs_list_recursive(
-                container_name=self.container_name,
-                prefix=blob_prefix or None,
+            existing_blobs = (
+                wasb_hook.get_blobs_list_recursive(
+                    container_name=self.container_name,
+                    prefix=blob_prefix or None,
+                )
+                or []
             )
-            existing_blobs = existing_blobs or []
             if blob_prefix:
                 prefix_str = blob_prefix.rstrip("/") + "/"
                 existing_blobs = [b.removeprefix(prefix_str) for b in existing_blobs]
-
             existing_blobs_set = set(existing_blobs)
-            filtered_files = []
-            seen_transformed = set()
 
-            for file in gcs_files:
-                transformed = self._transform_file_path(file)
-                if transformed not in existing_blobs_set and transformed not in seen_transformed:
-                    filtered_files.append(file)
-                    seen_transformed.add(transformed)
-                elif transformed in seen_transformed:
-                    self.log.warning(
-                        "Skipping duplicate file %s (transforms to %s)",
-                        file,
-                        transformed,
-                    )
+        filtered_files: list[str] = []
+        seen_transformed: set[str] = set()
+        for file in gcs_files:
+            transformed = self._transform_file_path(file)
+            if transformed in existing_blobs_set:
+                continue
+            if transformed in seen_transformed:
+                self.log.warning(
+                    "Skipping duplicate file %s (transforms to %s)",
+                    file,
+                    transformed,
+                )
+                continue
+            filtered_files.append(file)
+            seen_transformed.add(transformed)
 
-            gcs_files = filtered_files
+        gcs_files = filtered_files
 
         uploaded_blobs: list[str] = []
         if gcs_files:

providers/microsoft/azure/tests/unit/microsoft/azure/transfers/test_gcs_to_wasb.py

@@ -182,6 +182,28 @@ def test_flatten_structure(self, mock_gcs_hook, mock_wasb_hook):
         assert str(kw["blob_name"]).replace("\\", "/") == _expected_blob_paths(BLOB_PREFIX, "a.txt")[0]
         assert kw["overwrite"] is True
 
+    @mock.patch("airflow.providers.microsoft.azure.transfers.gcs_to_wasb.WasbHook")
+    @mock.patch("airflow.providers.microsoft.azure.transfers.gcs_to_wasb.GCSHook")
+    def test_execute_dedups_collisions_when_replace_true(self, mock_gcs_hook, mock_wasb_hook):
+        mock_gcs_hook.return_value.list.return_value = ["a/file.txt", "b/file.txt"]
+        mock_file = mock.Mock()
+        mock_file.name = "/tmp/local"
+        mock_gcs_hook.return_value.provide_file.return_value.__enter__ = mock.Mock(return_value=mock_file)
+        mock_gcs_hook.return_value.provide_file.return_value.__exit__ = mock.Mock(return_value=None)
+
+        op = GCSToAzureBlobStorageOperator(
+            task_id=TASK_ID,
+            gcs_bucket=GCS_BUCKET,
+            container_name=CONTAINER,
+            blob_prefix=BLOB_PREFIX,
+            replace=True,
+            flatten_structure=True,
+        )
+        result = op.execute(context=None)
+
+        assert _norm_paths(result) == _expected_blob_paths(BLOB_PREFIX, "file.txt")
+        mock_wasb_hook.return_value.load_file.assert_called_once()
+
     @mock.patch("airflow.providers.microsoft.azure.transfers.gcs_to_wasb.WasbHook")
     @mock.patch("airflow.providers.microsoft.azure.transfers.gcs_to_wasb.GCSHook")
     def test_execute_is_idempotent_on_retry(self, mock_gcs_hook, mock_wasb_hook):