feat(cdp): add microsoft teams template

posthog/cdp/templates/__init__.py

@@ -27,6 +27,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 .google_cloud_storage.template_google_cloud_storage import (
     template as google_cloud_storage,
     TemplateGoogleCloudStorageMigrator,
@@ -56,6 +57,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,62 @@
+from posthog.cdp.templates.hog_function_template import HogFunctionTemplate, HogFunctionSubTemplate, SUB_TEMPLATE_COMMON
+
+template: HogFunctionTemplate = HogFunctionTemplate(
+    status="beta",
+    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="""
+let res := fetch(inputs.webhookUrl, {
+    'body': {
+        'text': inputs.content
+    },
+    '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": "content",
+            "type": "string",
+            "label": "Content",
+            "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={
+                "content": "**{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={"content": "**{person.name}** responded to survey **{event.properties.$survey_name}**"},
+        ),
+    ],
+)

posthog/cdp/templates/microsoft_teams/test_template_microsoft_teams.py

@@ -0,0 +1,33 @@
+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://max.webhook.office.com/webhookb2/abcdefg@abcdefg/IncomingWebhook/abcdefg/abcdefg",
+            "content": "**[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://max.webhook.office.com/webhookb2/abcdefg@abcdefg/IncomingWebhook/abcdefg/abcdefg",
+                {
+                    "method": "POST",
+                    "headers": {
+                        "Content-Type": "application/json",
+                    },
+                    "body": {
+                        "text": "**[email protected]** triggered event: '$pageview'",
+                    },
+                },
+            )
+        )