Skip to content

Commit

Permalink
Add support for secure Mattermost notifications
Browse files Browse the repository at this point in the history
Apprise supports this but we need to surface the configuration part.
Keeping default to insecure connection to not break compatibility, but
perhaps worth reconsidering.
  • Loading branch information
Stanislav Ochotnický committed Jan 28, 2025
1 parent 10d2c75 commit 719aebf
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 719aebf

Please # to comment.