Skip to content

Commit

Permalink
Add support for secure Mattermost notifications (#16884)
Browse files Browse the repository at this point in the history
Co-authored-by: Stanislav Ochotnický <[email protected]>
  • Loading branch information
sochotnicky and Stanislav Ochotnický authored Jan 29, 2025
1 parent e441d14 commit bd807de
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 0 deletions.
5 changes: 5 additions & 0 deletions src/prefect/blocks/notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,10 @@ class MattermostWebhook(AbstractAppriseNotificationBlock):
description="The hostname of your Mattermost server.",
examples=["Mattermost.example.com"],
)
secure: bool = Field(
default=False,
description="Whether to use secure https connection.",
)

token: SecretStr = Field(
default=...,
Expand Down Expand Up @@ -621,6 +625,7 @@ def block_initialization(self) -> None:
channels=self.channels,
include_image=self.include_image,
port=self.port,
secure=self.secure,
).url() # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] incomplete type hints in apprise
)
self._start_apprise_client(url)
Expand Down
24 changes: 24 additions & 0 deletions tests/blocks/test_notifications.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,30 @@ async def test_notify_async(self):
body="test", title="", notify_type=PREFECT_NOTIFY_TYPE_DEFAULT
)

def test_notify_secure(self):
with patch("apprise.Apprise", autospec=True) as AppriseMock:
apprise_instance_mock = AppriseMock.return_value
apprise_instance_mock.async_notify = AsyncMock()

mm_block = MattermostWebhook(
hostname="example.com", token="token", secure=True, port=443
)

@flow
def test_flow():
mm_block.notify("test")

test_flow()

AppriseMock.assert_called_once()
apprise_instance_mock.add.assert_called_once_with(
f"mmosts://{mm_block.hostname}/{mm_block.token.get_secret_value()}/"
"?image=no&format=text&overflow=upstream"
)
apprise_instance_mock.async_notify.assert_called_once_with(
body="test", title="", notify_type=PREFECT_NOTIFY_TYPE_DEFAULT
)

def test_notify_sync(self):
with patch("apprise.Apprise", autospec=True) as AppriseMock:
apprise_instance_mock = AppriseMock.return_value
Expand Down

0 comments on commit bd807de

Please sign in to comment.