Add ack behavior

src/prefect/blocks/notifications.py

@@ -93,7 +93,12 @@ async def notify(
         subject: Optional[str] = None,
     ):
         if not self.allow_private_urls:
-            validate_restricted_url(self.url.get_secret_value())
+            try:
+                validate_restricted_url(self.url.get_secret_value())
+            except ValueError as exc:
+                if self._raise_on_failure:
+                    raise NotificationError(str(exc))
+                raise
 
         await super().notify(body, subject)
 

tests/blocks/test_notifications.py

@@ -6,6 +6,7 @@
 import pytest
 import respx
 
+from prefect.blocks.abstract import NotificationError
 from prefect.blocks.notifications import (
     PREFECT_NOTIFY_TYPE_DEFAULT,
     AppriseNotificationBlock,
@@ -126,6 +127,27 @@ async def test_notification_can_prevent_restricted_urls(
         with pytest.raises(ValueError, match=f"is not a valid URL.*{reason}"):
             await notification.notify(subject="example", body="example")
 
+    async def test_raises_on_url_validation_failure(self, block_class):
+        """
+        When within a raise_on_failure block, we want URL validation errors to be
+        wrapped and captured as NotificationErrors for reporting back to users.
+        """
+        block = block_class(url="https://127.0.0.1/foo/bar", allow_private_urls=False)
+
+        # outside of a raise_on_failure block, we get a ValueError directly
+        with pytest.raises(ValueError, match="not a valid URL") as captured:
+            await block.notify(subject="Test", body="Test")
+
+        # inside of a raise_on_failure block, we get a NotificationError
+        with block.raise_on_failure():
+            with pytest.raises(NotificationError) as captured:
+                await block.notify(subject="Test", body="Test")
+
+        assert captured.value.log == (
+            "'https://127.0.0.1/foo/bar' is not a valid URL.  It resolves to the "
+            "private address 127.0.0.1."
+        )
+
 
 class TestMattermostWebhook:
     async def test_notify_async(self):