add MattermostWebhook notification block

src/prefect/blocks/notifications.py

@@ -3,6 +3,7 @@
 
 import apprise
 from apprise import Apprise, AppriseAsset, NotifyType
+from apprise.plugins.NotifyMattermost import NotifyMattermost
 from apprise.plugins.NotifyOpsgenie import NotifyOpsgenie
 from apprise.plugins.NotifyPagerDuty import NotifyPagerDuty
 from apprise.plugins.NotifyTwilio import NotifyTwilio
@@ -400,3 +401,78 @@ def block_initialization(self) -> None:
             ).url()
         )
         self._start_apprise_client(url)
+
+
+class MattermostWebhook(AbstractAppriseNotificationBlock):
+    """
+    Enables sending notifications via a provided Mattermost webhook.
+    See [Apprise notify_Mattermost docs](https://github.com/caronc/apprise/wiki/Notify_Mattermost) # noqa
+
+
+    Examples:
+        Load a saved Mattermost webhook and send a message:
+        ```python
+        from prefect.blocks.notifications import MattermostWebhook
+
+        mattermost_webhook_block = MattermostWebhook.load("BLOCK_NAME")
+
+        mattermost_webhook_block.notify("Hello from Prefect!")
+        ```
+    """
+
+    _description = "Enables sending notifications via a provided Mattermost webhook."
+    _block_type_name = "Mattermost Webhook"
+    _block_type_slug = "mattermost-webhook"
+    _logo_url = "https://images.ctfassets.net/zscdif0zqppk/3mlbsJDAmK402ER1sf0zUF/a48ac43fa38f395dd5f56c6ed29f22bb/mattermost-logo-png-transparent.png?h=250"
+    _documentation_url = "https://docs.prefect.io/api-ref/prefect/blocks/notifications/#prefect.blocks.notifications.MattermostWebhook"
+
+    hostname: str = Field(
+        default=...,
+        description="The hostname of your Mattermost server.",
+        example="Mattermost.example.com",
+    )
+
+    token: SecretStr = Field(
+        default=...,
+        description="The token associated with your Mattermost webhook.",
+    )
+
+    botname: Optional[str] = Field(
+        title="Bot name",
+        default=None,
+        description="The name of the bot that will send the message.",
+    )
+
+    channels: Optional[List[str]] = Field(
+        default=None,
+        description="The channel(s) you wish to notify.",
+    )
+
+    include_image: bool = Field(
+        default=False,
+        description="Whether to include the Apprise status image in the message.",
+    )
+
+    path: Optional[str] = Field(
+        default=None,
+        description="An optional sub-path specification to append to the hostname.",
+    )
+
+    port: int = Field(
+        default=8065,
+        description="The port of your Mattermost server.",
+    )
+
+    def block_initialization(self) -> None:
+        url = SecretStr(
+            NotifyMattermost(
+                token=self.token.get_secret_value(),
+                fullpath=self.path,
+                host=self.hostname,
+                botname=self.botname,
+                channels=self.channels,
+                include_image=self.include_image,
+                port=self.port,
+            ).url()
+        )
+        self._start_apprise_client(url)

tests/blocks/test_notifications.py

@@ -8,6 +8,7 @@
 import prefect
 from prefect.blocks.notifications import (
     AppriseNotificationBlock,
+    MattermostWebhook,
     OpsgenieWebhook,
     PagerDutyWebHook,
     PrefectNotifyType,
@@ -89,6 +90,82 @@ def test_is_picklable(self, block_class: Type[AppriseNotificationBlock]):
         assert isinstance(unpickled, block_class)
 
 
+class TestMattermostWebhook:
+    async def test_notify_async(self):
+        with patch("apprise.Apprise", autospec=True) as AppriseMock:
+            reload_modules()
+
+            apprise_instance_mock = AppriseMock.return_value
+            apprise_instance_mock.async_notify = AsyncMock()
+
+            mm_block = MattermostWebhook(
+                hostname="example.com",
+                token="token",
+                include_image=True,
+            )
+            await mm_block.notify("test")
+
+            AppriseMock.assert_called_once()
+            apprise_instance_mock.add.assert_called_once_with(
+                f"mmost://{mm_block.hostname}/{mm_block.token.get_secret_value()}/"
+                "?image=yes&format=text&overflow=upstream&rto=4.0&cto=4.0&verify=yes"
+            )
+            apprise_instance_mock.async_notify.assert_awaited_once_with(
+                body="test", title=None, notify_type=PrefectNotifyType.DEFAULT
+            )
+
+    def test_notify_sync(self):
+        with patch("apprise.Apprise", autospec=True) as AppriseMock:
+            reload_modules()
+
+            apprise_instance_mock = AppriseMock.return_value
+            apprise_instance_mock.async_notify = AsyncMock()
+
+            mm_block = MattermostWebhook(hostname="example.com", token="token")
+            mm_block.notify("test")
+
+            AppriseMock.assert_called_once()
+            apprise_instance_mock.add.assert_called_once_with(
+                f"mmost://{mm_block.hostname}/{mm_block.token.get_secret_value()}/"
+                "?image=no&format=text&overflow=upstream&rto=4.0&cto=4.0&verify=yes"
+            )
+            apprise_instance_mock.async_notify.assert_called_once_with(
+                body="test", title=None, notify_type=PrefectNotifyType.DEFAULT
+            )
+
+    def test_notify_with_multiple_channels(self):
+        with patch("apprise.Apprise", autospec=True) as AppriseMock:
+            reload_modules()
+
+            apprise_instance_mock = AppriseMock.return_value
+            apprise_instance_mock.async_notify = AsyncMock()
+
+            mm_block = MattermostWebhook(
+                hostname="example.com",
+                token="token",
+                channels=["general", "death-metal-anonymous"],
+            )
+            mm_block.notify("test")
+
+            AppriseMock.assert_called_once()
+            apprise_instance_mock.add.assert_called_once_with(
+                f"mmost://{mm_block.hostname}/{mm_block.token.get_secret_value()}/"
+                "?image=no&format=text&overflow=upstream&rto=4.0&cto=4.0&verify=yes"
+                "&channel=death-metal-anonymous%2Cgeneral"
+            )
+
+            apprise_instance_mock.async_notify.assert_called_once_with(
+                body="test", title=None, notify_type=PrefectNotifyType.DEFAULT
+            )
+
+    def test_is_picklable(self):
+        reload_modules()
+        block = MattermostWebhook(token="token", hostname="example.com")
+        pickled = cloudpickle.dumps(block)
+        unpickled = cloudpickle.loads(pickled)
+        assert isinstance(unpickled, MattermostWebhook)
+
+
 class TestOpsgenieWebhook:
     API_KEY = "api_key"