feat(cdp): add microsoft teams template (#25652)

Co-authored-by: github-actions <41898282+github-actions[bot]@users.noreply.github.com>

posthog/cdp/templates/__init__.py

@@ -28,6 +28,7 @@
 from .activecampaign.template_activecampaign import template as activecampaign
 from .google_ads.template_google_ads import template as google_ads
 from .attio.template_attio import template as attio
+from .microsoft_teams.template_microsoft_teams import template as microsoft_teams
 from .klaviyo.template_klaviyo import template_user as klaviyo_user, template_event as klaviyo_event
 from .google_cloud_storage.template_google_cloud_storage import (
     template as google_cloud_storage,
@@ -62,6 +63,7 @@
     mailjet_create_contact,
     mailjet_update_contact_list,
     meta_ads,
+    microsoft_teams,
     posthog,
     rudderstack,
     salesforce_create,

posthog/cdp/templates/microsoft_teams/template_microsoft_teams.py

@@ -0,0 +1,83 @@
+from posthog.cdp.templates.hog_function_template import HogFunctionTemplate, HogFunctionSubTemplate, SUB_TEMPLATE_COMMON
+
+template: HogFunctionTemplate = HogFunctionTemplate(
+    status="free",
+    id="template-microsoft-teams",
+    name="Microsoft Teams",
+    description="Sends a message to a Microsoft Teams channel",
+    icon_url="/static/services/microsoft-teams.png",
+    category=["Customer Success"],
+    hog="""
+if (not match(inputs.webhookUrl, '^https://[^/]+.logic.azure.com:443/workflows/[^/]+/triggers/manual/paths/invoke?.*')) {
+    throw Error('Invalid URL. The URL should match the format: https://<region>.logic.azure.com:443/workflows/<workflowId>/triggers/manual/paths/invoke?...')
+}
+
+let res := fetch(inputs.webhookUrl, {
+    'body': {
+        'type': 'message',
+        'attachments': [
+            {
+                'contentType': 'application/vnd.microsoft.card.adaptive',
+                'contentUrl': null,
+                'content': {
+                    '$schema': 'http://adaptivecards.io/schemas/adaptive-card.json',
+                    'type': 'AdaptiveCard',
+                    'version': '1.2',
+                    'body': [
+                        {
+                            'type': 'TextBlock',
+                            'text': inputs.text
+                        }
+                    ]
+                }
+            }
+        ]
+    },
+    'method': 'POST',
+    'headers': {
+        'Content-Type': 'application/json'
+    }
+});
+
+if (res.status >= 400) {
+    throw Error(f'Failed to post message to Microsoft Teams: {res.status}: {res.body}');
+}
+""".strip(),
+    inputs_schema=[
+        {
+            "key": "webhookUrl",
+            "type": "string",
+            "label": "Webhook URL",
+            "description": "See this page on how to generate a Webhook URL: https://learn.microsoft.com/en-us/microsoftteams/platform/webhooks-and-connectors/how-to/add-incoming-webhook?tabs=newteams%2Cdotnet#create-an-incoming-webhook",
+            "secret": False,
+            "required": True,
+        },
+        {
+            "key": "text",
+            "type": "string",
+            "label": "Text",
+            "description": "(see https://learn.microsoft.com/en-us/microsoftteams/platform/webhooks-and-connectors/how-to/add-incoming-webhook?tabs=newteams%2Cdotnet#example)",
+            "default": "**{person.name}** triggered event: '{event.event}'",
+            "secret": False,
+            "required": True,
+        },
+    ],
+    sub_templates=[
+        HogFunctionSubTemplate(
+            id="early_access_feature_enrollment",
+            name="Post to Microsoft Teams on feature enrollment",
+            description="Posts a message to Microsoft Teams when a user enrolls or un-enrolls in an early access feature",
+            filters=SUB_TEMPLATE_COMMON["early_access_feature_enrollment"].filters,
+            inputs={
+                "text": "**{person.name}** {event.properties.$feature_enrollment ? 'enrolled in' : 'un-enrolled from'} the early access feature for '{event.properties.$feature_flag}'"
+            },
+        ),
+        HogFunctionSubTemplate(
+            id="survey_response",
+            name="Post to Microsoft Teams on survey response",
+            description="Posts a message to Microsoft Teams when a user responds to a survey",
+            filters=SUB_TEMPLATE_COMMON["survey_response"].filters,
+            inputs={"text": "**{person.name}** responded to survey **{event.properties.$survey_name}**"},
+        ),
+    ],
+)

posthog/cdp/templates/microsoft_teams/test_template_microsoft_teams.py

@@ -0,0 +1,74 @@
+import pytest
+from inline_snapshot import snapshot
+from posthog.cdp.templates.helpers import BaseHogFunctionTemplateTest
+from posthog.cdp.templates.microsoft_teams.template_microsoft_teams import template as template_microsoft_teams
+
+
+class TestTemplateMicrosoftTeams(BaseHogFunctionTemplateTest):
+    template = template_microsoft_teams
+
+    def _inputs(self, **kwargs):
+        inputs = {
+            "webhookUrl": "https://prod-180.westus.logic.azure.com:443/workflows/abc/triggers/manual/paths/invoke?api-version=2016-06-01",
+            "text": "**[email protected]** triggered event: '$pageview'",
+        }
+        inputs.update(kwargs)
+        return inputs
+
+    def test_function_works(self):
+        self.run_function(inputs=self._inputs())
+
+        assert self.get_mock_fetch_calls()[0] == snapshot(
+            (
+                "https://prod-180.westus.logic.azure.com:443/workflows/abc/triggers/manual/paths/invoke?api-version=2016-06-01",
+                {
+                    "method": "POST",
+                    "headers": {
+                        "Content-Type": "application/json",
+                    },
+                    "body": {
+                        "type": "message",
+                        "attachments": [
+                            {
+                                "contentType": "application/vnd.microsoft.card.adaptive",
+                                "contentUrl": None,
+                                "content": {
+                                    "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
+                                    "type": "AdaptiveCard",
+                                    "version": "1.2",
+                                    "body": [
+                                        {
+                                            "type": "TextBlock",
+                                            "text": "**[email protected]** triggered event: '$pageview'",
+                                        }
+                                    ],
+                                },
+                            }
+                        ],
+                    },
+                },
+            )
+        )
+
+    def test_only_allow_teams_url(self):
+        for url, allowed in [
+            [
+                "https://prod-180.westus.logic.azure.com:443/workflows/abc/triggers/manual/paths/invoke?api-version=2016-06-01",
+                True,
+            ],
+            ["https://webhook.site/def", False],
+            [
+                "https://webhook.site/def#https://prod-180.westus.logic.azure.com:443/workflows/abc/triggers/manual/paths/invoke?api-version=2016-06-01",
+                False,
+            ],
+        ]:
+            if allowed:
+                self.run_function(inputs=self._inputs(webhookUrl=url))
+                assert len(self.get_mock_fetch_calls()) == 1
+            else:
+                with pytest.raises(Exception) as e:
+                    self.run_function(inputs=self._inputs(webhookUrl=url))
+                assert (
+                    e.value.message  # type: ignore[attr-defined]
+                    == "Invalid URL. The URL should match the format: https://<region>.logic.azure.com:443/workflows/<workflowId>/triggers/manual/paths/invoke?..."
+                )