From 930388317a1da81d28b231fa616f9ce5940ab93d Mon Sep 17 00:00:00 2001
From: nate nowack <thrast36@gmail.com>
Date: Mon, 29 Jul 2024 09:37:03 -0500
Subject: [PATCH] Backport teams notif update for `2.x` (#14774)

---
 requirements.txt                    |  2 +-
 src/prefect/blocks/notifications.py | 30 ++++++++++++++--
 tests/blocks/test_notifications.py  | 55 ++++++++++++++++++++++++++++-
 3 files changed, 83 insertions(+), 4 deletions(-)

diff --git a/requirements.txt b/requirements.txt
index 7ff3e22fe6a8..ba5a41cc4ae9 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,7 +2,7 @@
 
 aiosqlite >= 0.17.0
 alembic >= 1.7.5, < 2.0.0
-apprise >= 1.1.0, < 2.0.0
+apprise >= 1.8.0, < 2.0.0
 asyncpg >= 0.23
 click >= 8.0, < 8.2
 cryptography >= 36.0.1
diff --git a/src/prefect/blocks/notifications.py b/src/prefect/blocks/notifications.py
index f1366ac71e65..8e0ce941732b 100644
--- a/src/prefect/blocks/notifications.py
+++ b/src/prefect/blocks/notifications.py
@@ -139,12 +139,38 @@ class MicrosoftTeamsWebhook(AppriseNotificationBlock):
     url: SecretStr = Field(
         ...,
         title="Webhook URL",
-        description="The Teams incoming webhook URL used to send notifications.",
+        description="The Microsoft Power Automate (Workflows) URL used to send notifications to Teams.",
         examples=[
-            "https://your-org.webhook.office.com/webhookb2/XXX/IncomingWebhook/YYY/ZZZ"
+            "https://prod-NO.LOCATION.logic.azure.com:443/workflows/WFID/triggers/manual/paths/invoke?sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=SIGNATURE"
         ],
     )
 
+    include_image: bool = Field(
+        default=True,
+        description="Include an image with the notification.",
+    )
+
+    wrap: bool = Field(
+        default=True,
+        description="Wrap the notification text.",
+    )
+
+    def block_initialization(self) -> None:
+        """see https://github.com/caronc/apprise/pull/1172"""
+        from apprise.plugins.workflows import NotifyWorkflows
+
+        if not (
+            parsed_url := NotifyWorkflows.parse_native_url(self.url.get_secret_value())
+        ):
+            raise ValueError("Invalid Microsoft Teams Workflow URL provided.")
+
+        parsed_url |= {  # novermin
+            "include_image": self.include_image,
+            "wrap": self.wrap,
+        }
+
+        self._start_apprise_client(SecretStr(NotifyWorkflows(**parsed_url).url()))
+
 
 class PagerDutyWebHook(AbstractAppriseNotificationBlock):
     """
diff --git a/tests/blocks/test_notifications.py b/tests/blocks/test_notifications.py
index 6808f1c8c618..6da718c5c2ee 100644
--- a/tests/blocks/test_notifications.py
+++ b/tests/blocks/test_notifications.py
@@ -14,6 +14,7 @@
     CustomWebhookNotificationBlock,
     DiscordWebhook,
     MattermostWebhook,
+    MicrosoftTeamsWebhook,
     OpsgenieWebhook,
     PagerDutyWebHook,
     SendgridEmail,
@@ -39,7 +40,12 @@ def reload_modules():
 
 # A list of the notification classes Pytest should use as parameters to each method in TestAppriseNotificationBlock
 notification_classes = sorted(
-    AppriseNotificationBlock.__subclasses__(), key=lambda cls: cls.__name__
+    [
+        cls
+        for cls in AppriseNotificationBlock.__subclasses__()
+        if cls != MicrosoftTeamsWebhook
+    ],
+    key=lambda cls: cls.__name__,
 )
 
 
@@ -702,3 +708,50 @@ def test_is_picklable(self):
         pickled = cloudpickle.dumps(block)
         unpickled = cloudpickle.loads(pickled)
         assert isinstance(unpickled, SendgridEmail)
+
+
+class TestMicrosoftTeamsWebhook:
+    SAMPLE_URL = "https://prod-NO.LOCATION.logic.azure.com:443/workflows/WFID/triggers/manual/paths/invoke?sp=%2Ftriggers%2Fmanual%2Frun&sv=1.0&sig=SIGNATURE"
+
+    async def test_notify_async(self):
+        with patch("apprise.Apprise", autospec=True) as AppriseMock:
+            apprise_instance_mock = AppriseMock.return_value
+            apprise_instance_mock.async_notify = AsyncMock()
+
+            block = MicrosoftTeamsWebhook(url=self.SAMPLE_URL)
+            await block.notify("test")
+
+            AppriseMock.assert_called_once()
+            apprise_instance_mock.add.assert_called_once_with(
+                "workflow://prod-NO.LOCATION.logic.azure.com:443/WFID/SIGNATURE/"
+                "?image=yes&wrap=yes"
+                "&format=markdown&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=PREFECT_NOTIFY_TYPE_DEFAULT
+            )
+
+    def test_notify_sync(self):
+        with patch("apprise.Apprise", autospec=True) as AppriseMock:
+            apprise_instance_mock = AppriseMock.return_value
+            apprise_instance_mock.async_notify = AsyncMock()
+
+            block = MicrosoftTeamsWebhook(url=self.SAMPLE_URL)
+
+            block.notify("test")
+
+            AppriseMock.assert_called_once()
+            apprise_instance_mock.add.assert_called_once_with(
+                "workflow://prod-NO.LOCATION.logic.azure.com:443/WFID/SIGNATURE/"
+                "?image=yes&wrap=yes"
+                "&format=markdown&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=PREFECT_NOTIFY_TYPE_DEFAULT
+            )
+
+    def test_is_picklable(self):
+        block = MicrosoftTeamsWebhook(url=self.SAMPLE_URL)
+        pickled = cloudpickle.dumps(block)
+        unpickled = cloudpickle.loads(pickled)
+        assert isinstance(unpickled, MicrosoftTeamsWebhook)