Revert "GCSHook: Log NotFound error instead of raise on blob delete"

This reverts commit 6ba1f57eb153e0d26c9420dcff1ee4c7f06724a2.

Author: tuzonghua

providers/google/src/airflow/providers/google/cloud/hooks/gcs.py

@@ -715,16 +715,12 @@ def delete(self, bucket_name: str, object_name: str) -> None:
         client = self.get_conn()
         bucket = client.bucket(bucket_name)
         blob = bucket.blob(blob_name=object_name)
+        blob.delete()
+        get_hook_lineage_collector().add_input_asset(
+            context=self, scheme="gs", asset_kwargs={"bucket": bucket.name, "key": blob.name}
+        )
 
-        self.log.info("Deleting blob %s", object_name)
-        try:
-            blob.delete()
-            get_hook_lineage_collector().add_input_asset(
-                context=self, scheme="gs", asset_kwargs={"bucket": bucket.name, "key": blob.name}
-            )
-            self.log.info("Blob %s has been deleted", object_name)
-        except NotFound:
-            self.log.info("Blob %s does not exist", object_name)
+        self.log.info("Blob %s deleted.", object_name)
 
     def get_bucket(self, bucket_name: str) -> storage.Bucket:
         """

providers/google/tests/unit/google/cloud/hooks/test_gcs.py

@@ -528,26 +528,21 @@ def test_rewrite_exposes_lineage(self, mock_service, hook_lineage_collector):
 
     @mock.patch("google.cloud.storage.Bucket")
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
-    def test_delete(self, mock_service, mock_bucket, caplog):
+    def test_delete(self, mock_service, mock_bucket):
         test_bucket = "test_bucket"
         test_object = "test_object"
         blob_to_be_deleted = storage.Blob(name=test_object, bucket=mock_bucket)
 
-        bucket_method = mock_service.return_value.bucket
-        blob = bucket_method.return_value.blob
-        delete_method = blob.return_value.delete
+        get_bucket_method = mock_service.return_value.get_bucket
+        get_blob_method = get_bucket_method.return_value.get_blob
+        delete_method = get_blob_method.return_value.delete
         delete_method.return_value = blob_to_be_deleted
 
-        with caplog.at_level(logging.INFO):
-            response = self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object)
+        response = self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object)
         assert response is None
-        bucket_method.assert_called_once_with(test_bucket)
-        blob.assert_called_once_with(blob_name=test_object)
-        delete_method.assert_called_once()
-        assert "Blob test_object has been deleted" in caplog.text
 
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
-    def test_delete_nonexisting_object(self, mock_service, caplog):
+    def test_delete_nonexisting_object(self, mock_service):
         test_bucket = "test_bucket"
         test_object = "test_object"
 
@@ -556,12 +551,8 @@ def test_delete_nonexisting_object(self, mock_service, caplog):
         delete_method = blob.return_value.delete
         delete_method.side_effect = NotFound(message="Not Found")
 
-        with caplog.at_level(logging.INFO):
+        with pytest.raises(NotFound):
             self.gcs_hook.delete(bucket_name=test_bucket, object_name=test_object)
-        bucket_method.assert_called_once_with(test_bucket)
-        blob.assert_called_once_with(blob_name=test_object)
-        delete_method.assert_called_once()
-        assert "Blob test_object does not exist" in caplog.text
 
     @mock.patch(GCS_STRING.format("GCSHook.get_conn"))
     def test_delete_exposes_lineage(self, mock_service, hook_lineage_collector):