✨ feat(aci): add pagerduty/opsegnie migration utils

src/sentry/workflow_engine/typings/notification_action.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from typing import Any, ClassVar
 
+from sentry.integrations.opsgenie.client import OPSGENIE_DEFAULT_PRIORITY
+from sentry.integrations.pagerduty.client import PAGERDUTY_DEFAULT_SEVERITY
 from sentry.notifications.models.notificationaction import ActionTarget
 from sentry.utils.registry import Registry
 from sentry.workflow_engine.models.action import Action
@@ -11,8 +13,18 @@
 EXCLUDED_ACTION_DATA_KEYS = ["uuid", "id"]
 
 
+@dataclass
+class FieldMapping:
+    """FieldMapping is a class that represents the mapping of a target field to a source field."""
+
+    source_field: str
+    default_value: Any = None
+
+
 class BaseActionTranslator(ABC):
     action_type: ClassVar[Action.Type]
+    # Represents the mapping of a target field to a source field {target_field: FieldMapping}
+    field_mappings: ClassVar[dict[str, FieldMapping]] = {}
 
     def __init__(self, action: dict[str, Any]):
         self.action = action
@@ -69,10 +81,19 @@ def get_sanitized_data(self) -> dict[str, Any]:
         Otherwise, remove excluded keys
         """
         if self.blob_type:
-            # Convert to dataclass if blob type is specified
-            blob_instance = self.blob_type(
-                **{k.name: self.action.get(k.name, "") for k in dataclasses.fields(self.blob_type)}
-            )
+            mapped_data = {}
+            for field in dataclasses.fields(self.blob_type):
+                mapping = self.field_mappings.get(field.name)
+                # If a mapping is specified, use the source field value or default value
+                if mapping:
+                    source_field = mapping.source_field
+                    value = self.action.get(source_field, mapping.default_value)
+                # Otherwise, use the field value
+                else:
+                    value = self.action.get(field.name, "")
+                mapped_data[field.name] = value
+
+            blob_instance = self.blob_type(**mapped_data)
             return dataclasses.asdict(blob_instance)
         else:
             # Remove excluded keys and required fields
@@ -168,6 +189,70 @@ def target_display(self) -> str | None:
         return self.action.get("channel")
 
 
+@issue_alert_action_translator_registry.register(
+    "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction"
+)
+class PagerDutyActionTranslator(BaseActionTranslator):
+    action_type = Action.Type.PAGERDUTY
+    field_mappings = {
+        "priority": FieldMapping(
+            source_field="severity", default_value=str(PAGERDUTY_DEFAULT_SEVERITY)
+        )
+    }
+
+    @property
+    def required_fields(self) -> list[str]:
+        return ["account", "service"]
+
+    @property
+    def target_type(self) -> ActionTarget:
+        return ActionTarget.SPECIFIC
+
+    @property
+    def integration_id(self) -> Any | None:
+        return self.action.get("account")
+
+    @property
+    def target_identifier(self) -> str | None:
+        return self.action.get("service")
+
+    @property
+    def blob_type(self) -> type["DataBlob"]:
+        return OnCallDataBlob
+
+
+@issue_alert_action_translator_registry.register(
+    "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction"
+)
+class OpsgenieActionTranslator(BaseActionTranslator):
+    action_type = Action.Type.OPSGENIE
+    field_mappings = {
+        "priority": FieldMapping(
+            source_field="priority", default_value=str(OPSGENIE_DEFAULT_PRIORITY)
+        )
+    }
+
+    @property
+    def required_fields(self) -> list[str]:
+        return ["account", "team"]
+
+    @property
+    def target_type(self) -> ActionTarget:
+        return ActionTarget.SPECIFIC
+
+    @property
+    def integration_id(self) -> Any | None:
+        return self.action.get("account")
+
+    @property
+    def target_identifier(self) -> str | None:
+        return self.action.get("team")
+
+    @property
+    def blob_type(self) -> type["DataBlob"]:
+        return OnCallDataBlob
+
+
 @dataclass
 class DataBlob:
     """DataBlob is a generic type that represents the data blob for a notification action."""
@@ -190,3 +275,10 @@ class DiscordDataBlob(DataBlob):
     """
 
     tags: str = ""
+
+
+@dataclass
+class OnCallDataBlob(DataBlob):
+    """OnCallDataBlob is a specific type that represents the data blob for a PagerDuty or Opsgenie notification action."""
+
+    priority: str = ""

tests/sentry/workflow_engine/migration_helpers/test_migrate_rule_action.py

@@ -44,8 +44,13 @@ def assert_action_data_blob(
         # If we have a blob type, verify the data matches the blob structure
         if translator.blob_type:
             for field in translator.blob_type.__dataclass_fields__:
-                if field not in exclude_keys:
-                    # Action should always have the field, it can be empty
+                mapping = translator.field_mappings.get(field)
+                if mapping:
+                    # For mapped fields, check against the source field with default value
+                    source_value = compare_dict.get(mapping.source_field, mapping.default_value)
+                    assert action.data.get(field) == source_value
+                else:
+                    # For unmapped fields, check directly with empty string default
                     assert action.data.get(field) == compare_dict.get(field, "")
             # Ensure no extra fields
             assert set(action.data.keys()) == {
@@ -376,3 +381,115 @@ def test_msteams_action_migration_malformed(self, mock_logger):
                 "missing_fields": ["channel_id", "channel"],
             },
         )
+
+    def test_pagerduty_action_migration(self):
+        action_data = [
+            {
+                "account": "123456",
+                "id": "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction",
+                "service": "91919",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+            },
+            {
+                "account": "999999",
+                "service": "19191",
+                "id": "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction",
+                "uuid": "9a8b7c6d-5e4f-3a2b-1c0d-9a8b7c6d5e4f",
+                "severity": "warning",
+            },
+            {
+                "account": "77777",
+                "service": "57436",
+                "severity": "info",
+                "id": "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+            },
+        ]
+
+        actions = build_notification_actions_from_rule_data_actions(action_data)
+        assert len(actions) == len(action_data)
+
+        # Verify that default value is used when severity is not provided
+        assert actions[0].data["priority"] == "default"
+        # Verify that severity is mapped to priority
+        assert actions[1].data["priority"] == "warning"
+        assert actions[2].data["priority"] == "info"
+
+        self.assert_actions_migrated_correctly(actions, action_data, "account", "service", None)
+
+    def test_pagerduty_action_migration_malformed(self):
+        action_data = [
+            # Missing required fields
+            {
+                "account": "123456",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+                "id": "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction",
+            },
+            {
+                "account": "123456",
+                "service": "91919",
+                "id": "sentry.integrations.pagerduty.notify_action.PagerDutyNotifyServiceAction",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+            },
+        ]
+
+        actions = build_notification_actions_from_rule_data_actions(action_data)
+        assert len(actions) == 1
+
+        self.assert_actions_migrated_correctly(actions, action_data[1:], "account", "service", None)
+
+    def test_opsgenie_action_migration(self):
+        action_data = [
+            {
+                "account": "11111",
+                "id": "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction",
+                "team": "2323213-bbbbbuuufffooobottt",
+                "uuid": "87654321-0987-6543-2109-876543210987",
+            },
+            {
+                "account": "123456",
+                "team": "1234-bufo-bot",
+                "priority": "P2",
+                "id": "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+            },
+            {
+                "account": "999999",
+                "team": "1234-bufo-bot-2",
+                "priority": "P3",
+                "id": "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction",
+                "uuid": "01234567-89ab-cdef-0123-456789abcdef",
+            },
+        ]
+
+        actions = build_notification_actions_from_rule_data_actions(action_data)
+        assert len(actions) == len(action_data)
+
+        # Verify that default value is used when priority is not provided
+        assert actions[0].data["priority"] == "P3"
+        # Verify that priority is mapped to priority
+        assert actions[1].data["priority"] == "P2"
+        assert actions[2].data["priority"] == "P3"
+
+        self.assert_actions_migrated_correctly(actions, action_data, "account", "team", None)
+
+    def test_opsgenie_action_migration_malformed(self):
+        action_data = [
+            # Missing required fields
+            {
+                "account": "123456",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+                "id": "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction",
+            },
+            {
+                "account": "123456",
+                "team": "1234-bufo-bot",
+                "id": "sentry.integrations.opsgenie.notify_action.OpsgenieNotifyTeamAction",
+                "uuid": "12345678-90ab-cdef-0123-456789abcdef",
+            },
+        ]
+
+        actions = build_notification_actions_from_rule_data_actions(action_data)
+        assert len(actions) == 1
+
+        self.assert_actions_migrated_correctly(actions, action_data[1:], "account", "team", None)