New Cloud Agent

src/sentry/rules/actions/notify_event_service.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Generator, Sequence
+from collections.abc import Generator, Mapping, Sequence
 from typing import Any
 
 from django import forms
@@ -30,6 +30,10 @@
 PLUGINS_WITH_FIRST_PARTY_EQUIVALENTS = ["PagerDuty", "Slack", "Opsgenie"]
 
 
+def _as_mapping(value: Any) -> Mapping[str, Any]:
+    return value if isinstance(value, Mapping) else {}
+
+
 def build_incident_attachment(
     alert_context: AlertContext,
     metric_issue_context: MetricIssueContext,
@@ -103,21 +107,28 @@ def find_alert_rule_action_ui_component(app_platform_event: AppPlatformEvent) ->
     Loop through the triggers for the alert rule event. For each trigger, check
     if an action is an alert rule UI Component
     """
-    triggers = (
-        getattr(app_platform_event, "data", {})
-        .get("metric_alert", {})
-        .get("alert_rule", {})
-        .get("triggers", [])
-    )
-
-    actions = [
-        action
-        for trigger in triggers
-        for action in trigger.get("actions", {})
-        if (action.get("type") == "sentry_app" and action.get("settings") is not None)
-    ]
+    data = _as_mapping(getattr(app_platform_event, "data", None))
+    metric_alert = _as_mapping(data.get("metric_alert"))
+    alert_rule = _as_mapping(metric_alert.get("alert_rule"))
+
+    triggers = alert_rule.get("triggers")
+    if not isinstance(triggers, Sequence) or isinstance(triggers, (str, bytes)):
+        triggers = []
+
+    actions = []
+    for trigger in triggers:
+        if not isinstance(trigger, Mapping):
+            continue
+        action_configs = trigger.get("actions") or []
+        if not isinstance(action_configs, Sequence) or isinstance(action_configs, (str, bytes)):
+            continue
+        for action in action_configs:
+            if not isinstance(action, Mapping):
+                continue
+            if action.get("type") == "sentry_app" and action.get("settings") is not None:
+                actions.append(action)
 
-    return bool(len(actions))
+    return bool(actions)
 
 
 class NotifyEventServiceForm(forms.Form):

tests/sentry/rules/actions/test_notify_event_service.py

@@ -2,10 +2,13 @@
 
 from django.utils import timezone
 
-from sentry.rules.actions.notify_event_service import NotifyEventServiceAction
+from sentry.rules.actions.notify_event_service import (
+    NotifyEventServiceAction,
+    find_alert_rule_action_ui_component,
+)
 from sentry.sentry_apps.tasks.sentry_apps import notify_sentry_app
 from sentry.silo.base import SiloMode
-from sentry.testutils.cases import RuleTestCase
+from sentry.testutils.cases import RuleTestCase, TestCase
 from sentry.testutils.silo import assume_test_silo_mode
 from sentry.testutils.skips import requires_snuba
 
@@ -90,3 +93,31 @@ def test_sentry_app_installed(self) -> None:
 
         results = rule.get_services()
         assert len(results) == 0
+
+
+class FindAlertRuleActionUiComponentTest(TestCase):
+    def test_returns_false_when_metric_alert_is_null(self) -> None:
+        app_platform_event = MagicMock()
+        app_platform_event.data = {
+            "metric_alert": None,
+        }
+
+        assert find_alert_rule_action_ui_component(app_platform_event) is False
+
+    def test_detects_ui_component_action(self) -> None:
+        app_platform_event = MagicMock()
+        app_platform_event.data = {
+            "metric_alert": {
+                "alert_rule": {
+                    "triggers": [
+                        {
+                            "actions": [
+                                {"type": "sentry_app", "settings": {"foo": "bar"}},
+                            ]
+                        }
+                    ]
+                }
+            }
+        }
+
+        assert find_alert_rule_action_ui_component(app_platform_event) is True