Do not catch too broad exception in WasbHook.delete_container (#36034)

GitOrigin-RevId: 3f354c0c92870e4e72ad536abd1c9cf49462b206

Author: Taragolis

airflow/providers/microsoft/azure/hooks/wasb.py

@@ -487,9 +487,9 @@ def delete_container(self, container_name: str) -> None:
             self._get_container_client(container_name).delete_container()
             self.log.info("Deleted container: %s", container_name)
         except ResourceNotFoundError:
-            self.log.info("Unable to delete container %s (not found)", container_name)
-        except:
-            self.log.info("Error deleting container: %s", container_name)
+            self.log.warning("Unable to delete container %s (not found)", container_name)
+        except Exception:
+            self.log.error("Error deleting container: %s", container_name)
             raise
 
     def delete_blobs(self, container_name: str, *blobs, **kwargs) -> None:

tests/providers/microsoft/azure/hooks/test_wasb.py

@@ -20,6 +20,7 @@
 from unittest import mock
 
 import pytest
+from azure.core.exceptions import ResourceNotFoundError
 from azure.storage.blob import BlobServiceClient
 from azure.storage.blob._models import BlobProperties
 
@@ -518,6 +519,26 @@ def test_delete_container(self, mocked_blob_service_client):
         mocked_container_client.assert_called_once_with("mycontainer")
         mocked_container_client.return_value.delete_container.assert_called()
 
+    @pytest.mark.parametrize("exc", [ValueError, RuntimeError])
+    def test_delete_container_generic_exception(self, exc: type[Exception], caplog):
+        hook = WasbHook(wasb_conn_id=self.azure_shared_key_test)
+        with mock.patch.object(WasbHook, "_get_container_client") as m:
+            m.return_value.delete_container.side_effect = exc("FakeException")
+            caplog.clear()
+            caplog.set_level("ERROR")
+            with pytest.raises(exc, match="FakeException"):
+                hook.delete_container("mycontainer")
+            assert "Error deleting container: mycontainer" in caplog.text
+
+    def test_delete_container_resource_not_found(self, caplog):
+        hook = WasbHook(wasb_conn_id=self.azure_shared_key_test)
+        with mock.patch.object(WasbHook, "_get_container_client") as m:
+            m.return_value.delete_container.side_effect = ResourceNotFoundError("FakeException")
+            caplog.clear()
+            caplog.set_level("WARNING")
+            hook.delete_container("mycontainer")
+            assert "Unable to delete container mycontainer (not found)" in caplog.text
+
     @mock.patch.object(WasbHook, "delete_blobs")
     def test_delete_single_blob(self, delete_blobs, mocked_blob_service_client):
         hook = WasbHook(wasb_conn_id=self.azure_shared_key_test)