diff --git a/src/prefect/blocks/notifications.py b/src/prefect/blocks/notifications.py index 9bc79ca6630b..4970a2b005b3 100644 --- a/src/prefect/blocks/notifications.py +++ b/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) diff --git a/tests/blocks/test_notifications.py b/tests/blocks/test_notifications.py index 07ce3040a5e9..8cf010ce63b4 100644 --- a/tests/blocks/test_notifications.py +++ b/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):