feat(ACI): Create shared test base class for workflow engine serializers

tests/sentry/incidents/serializers/test_workflow_engine_action.py

@@ -3,46 +3,19 @@
     WorkflowEngineActionSerializer,
 )
 from sentry.incidents.models.alert_rule import AlertRuleTriggerAction
-from sentry.testutils.cases import TestCase
-from sentry.testutils.helpers.datetime import freeze_time
 from sentry.workflow_engine.migration_helpers.alert_rule import (
-    migrate_alert_rule,
     migrate_metric_action,
     migrate_metric_data_conditions,
 )
+from tests.sentry.incidents.serializers.test_workflow_engine_base import TestWorklowEngineSerializer
 
 
-@freeze_time("2018-12-11 03:21:34")
-class TestActionSerializer(TestCase):
-    def setUp(self) -> None:
-        self.alert_rule = self.create_alert_rule()
-        self.critical_trigger = self.create_alert_rule_trigger(
-            alert_rule=self.alert_rule, label="critical"
-        )
-        self.critical_trigger_action = self.create_alert_rule_trigger_action(
-            alert_rule_trigger=self.critical_trigger
-        )
-        migrate_alert_rule(self.alert_rule)
-        migrate_metric_data_conditions(self.critical_trigger)
-        self.action, _, _ = migrate_metric_action(self.critical_trigger_action)
-
-        self.expected = {
-            "id": str(self.critical_trigger_action.id),
-            "alertRuleTriggerId": str(self.critical_trigger.id),
-            "type": "email",
-            "targetType": "user",
-            "targetIdentifier": str(self.user.id),
-            "inputChannelId": None,
-            "integrationId": None,
-            "sentryAppId": None,
-            "dateCreated": self.action.date_added,
-            "desc": f"Send a notification to {self.user.email}",
-            "priority": self.action.data.get("priority"),
-        }
-
+class TestActionSerializer(TestWorklowEngineSerializer):
     def test_simple(self) -> None:
-        serialized_action = serialize(self.action, self.user, WorkflowEngineActionSerializer())
-        assert serialized_action == self.expected
+        serialized_action = serialize(
+            self.critical_action, self.user, WorkflowEngineActionSerializer()
+        )
+        assert serialized_action == self.expected_critical_action[0]
 
     def test_warning_trigger(self) -> None:
         """
@@ -63,7 +36,7 @@ def test_warning_trigger(self) -> None:
         serialized_action = serialize(
             self.warning_action, self.user, WorkflowEngineActionSerializer()
         )
-        warning_expected = self.expected.copy()
+        warning_expected = self.expected_critical_action[0].copy()
         warning_expected["id"] = str(self.warning_trigger_action.id)
         warning_expected["alertRuleTriggerId"] = str(self.warning_trigger.id)
         warning_expected["targetType"] = "team"
@@ -103,7 +76,7 @@ def test_sentry_app_action(self) -> None:
         serialized_action = serialize(
             self.sentry_app_action, self.user, WorkflowEngineActionSerializer()
         )
-        sentry_app_expected = self.expected.copy()
+        sentry_app_expected = self.expected_critical_action[0].copy()
         sentry_app_expected["type"] = "sentry_app"
         sentry_app_expected["id"] = str(self.sentry_app_trigger_action.id)
         sentry_app_expected["alertRuleTriggerId"] = str(self.sentry_app_trigger.id)
@@ -137,7 +110,7 @@ def test_slack_action(self) -> None:
         serialized_action = serialize(
             self.slack_action, self.user, WorkflowEngineActionSerializer()
         )
-        slack_expected = self.expected.copy()
+        slack_expected = self.expected_critical_action[0].copy()
         slack_expected["type"] = self.integration.provider
         slack_expected["id"] = str(self.slack_trigger_action.id)
         slack_expected["alertRuleTriggerId"] = str(self.slack_trigger.id)

tests/sentry/incidents/serializers/test_workflow_engine_base.py

@@ -0,0 +1,133 @@
+from django.utils import timezone
+
+from sentry.incidents.models.alert_rule import AlertRuleStatus, AlertRuleThresholdType
+from sentry.incidents.models.incident import IncidentTrigger, TriggerStatus
+from sentry.models.groupopenperiod import GroupOpenPeriod
+from sentry.testutils.cases import TestCase
+from sentry.testutils.helpers.datetime import freeze_time
+from sentry.types.group import PriorityLevel
+from sentry.workflow_engine.migration_helpers.alert_rule import (
+    migrate_alert_rule,
+    migrate_metric_action,
+    migrate_metric_data_conditions,
+    migrate_resolve_threshold_data_condition,
+)
+from sentry.workflow_engine.models import ActionGroupStatus
+
+
+@freeze_time("2024-12-11 03:21:34")
+class TestWorklowEngineSerializer(TestCase):
+    def setUp(self) -> None:
+        self.now = timezone.now()
+        self.alert_rule = self.create_alert_rule()
+        self.critical_trigger = self.create_alert_rule_trigger(
+            alert_rule=self.alert_rule, label="critical"
+        )
+        self.critical_trigger_action = self.create_alert_rule_trigger_action(
+            alert_rule_trigger=self.critical_trigger
+        )
+        _, _, _, self.detector, _, _, _, _ = migrate_alert_rule(self.alert_rule)
+        self.critical_detector_trigger, _ = migrate_metric_data_conditions(self.critical_trigger)
+
+        self.critical_action, _, _ = migrate_metric_action(self.critical_trigger_action)
+        self.resolve_trigger_data_condition = migrate_resolve_threshold_data_condition(
+            self.alert_rule
+        )
+        self.expected_critical_action = [
+            {
+                "id": str(self.critical_trigger_action.id),
+                "alertRuleTriggerId": str(self.critical_trigger.id),
+                "type": "email",
+                "targetType": "user",
+                "targetIdentifier": str(self.user.id),
+                "inputChannelId": None,
+                "integrationId": None,
+                "sentryAppId": None,
+                "dateCreated": self.critical_trigger_action.date_added,
+                "desc": f"Send a notification to {self.user.email}",
+                "priority": self.critical_action.data.get("priority"),
+            }
+        ]
+        self.expected_triggers = [
+            {
+                "id": str(self.critical_trigger.id),
+                "alertRuleId": str(self.alert_rule.id),
+                "label": "critical",
+                "thresholdType": AlertRuleThresholdType.ABOVE.value,
+                "alertThreshold": self.critical_detector_trigger.comparison,
+                "resolveThreshold": AlertRuleThresholdType.BELOW,
+                "dateCreated": self.critical_trigger.date_added,
+                "actions": self.expected_critical_action,
+            },
+        ]
+
+        self.expected = {
+            "id": str(self.alert_rule.id),
+            "name": self.detector.name,
+            "organizationId": self.detector.project.organization_id,
+            "status": AlertRuleStatus.PENDING.value,
+            "query": self.alert_rule.snuba_query.query,
+            "aggregate": self.alert_rule.snuba_query.aggregate,
+            "timeWindow": self.alert_rule.snuba_query.time_window,
+            "resolution": self.alert_rule.snuba_query.resolution,
+            "thresholdPeriod": self.detector.config.get("thresholdPeriod"),
+            "triggers": self.expected_triggers,
+            "projects": [self.project.slug],
+            "owner": self.detector.owner_user_id,
+            "dateModified": self.detector.date_updated,
+            "dateCreated": self.detector.date_added,
+            "createdBy": {},
+            "description": self.detector.description or "",
+            "detectionType": self.detector.type,
+        }
+
+    def add_warning_trigger(self) -> None:
+        self.warning_trigger = self.create_alert_rule_trigger(
+            alert_rule=self.alert_rule, label="warning"
+        )
+        self.warning_trigger_action = self.create_alert_rule_trigger_action(
+            alert_rule_trigger=self.warning_trigger
+        )
+        self.warning_detector_trigger, _ = migrate_metric_data_conditions(self.warning_trigger)
+        self.warning_action, _, _ = migrate_metric_action(self.warning_trigger_action)
+        self.expected_warning_action = [
+            {
+                "id": str(self.warning_trigger_action.id),
+                "alertRuleTriggerId": str(self.warning_trigger.id),
+                "type": "email",
+                "targetType": "user",
+                "targetIdentifier": str(self.user.id),
+                "inputChannelId": None,
+                "integrationId": None,
+                "sentryAppId": None,
+                "dateCreated": self.warning_trigger_action.date_added,
+                "desc": f"Send a notification to {self.user.email}",
+                "priority": self.warning_action.data.get("priority"),
+            }
+        ]
+        self.expected_warning_trigger = {
+            "id": str(self.warning_trigger.id),
+            "alertRuleId": str(self.alert_rule.id),
+            "label": "warning",
+            "thresholdType": AlertRuleThresholdType.ABOVE.value,
+            "alertThreshold": self.critical_detector_trigger.comparison,
+            "resolveThreshold": AlertRuleThresholdType.BELOW,
+            "dateCreated": self.critical_trigger.date_added,
+            "actions": self.expected_warning_action,
+        }
+        self.expected_triggers.append(self.expected_warning_trigger)
+
+    def add_incident_data(self) -> None:
+        self.incident = self.create_incident(alert_rule=self.alert_rule, date_started=self.now)
+        IncidentTrigger.objects.create(
+            incident=self.incident,
+            alert_rule_trigger=self.critical_trigger,
+            status=TriggerStatus.ACTIVE.value,
+        )
+
+        self.group.priority = PriorityLevel.HIGH
+        self.group.save()
+        ActionGroupStatus.objects.create(action=self.critical_action, group=self.group)
+        GroupOpenPeriod.objects.create(
+            group=self.group, project=self.detector.project, date_started=self.incident.date_started
+        )

tests/sentry/incidents/serializers/test_workflow_engine_data_condition.py

@@ -3,71 +3,20 @@
     WorkflowEngineDataConditionSerializer,
 )
 from sentry.incidents.endpoints.utils import translate_data_condition_type
-from sentry.incidents.models.alert_rule import (
-    AlertRule,
-    AlertRuleThresholdType,
-    AlertRuleTrigger,
-    AlertRuleTriggerAction,
-)
-from sentry.testutils.cases import TestCase
-from sentry.testutils.helpers.datetime import freeze_time
+from sentry.incidents.models.alert_rule import AlertRule, AlertRuleTrigger, AlertRuleTriggerAction
 from sentry.workflow_engine.migration_helpers.alert_rule import (
     migrate_alert_rule,
     migrate_metric_action,
     migrate_metric_data_conditions,
     migrate_resolve_threshold_data_condition,
 )
+from tests.sentry.incidents.serializers.test_workflow_engine_base import TestWorklowEngineSerializer
 
 
-@freeze_time("2018-12-11 03:21:34")
-class TestDataConditionSerializer(TestCase):
+class TestDataConditionSerializer(TestWorklowEngineSerializer):
     def setUp(self) -> None:
-        self.alert_rule = self.create_alert_rule()
-        self.critical_trigger = self.create_alert_rule_trigger(
-            alert_rule=self.alert_rule, label="critical"
-        )
-        self.critical_trigger_action = self.create_alert_rule_trigger_action(
-            alert_rule_trigger=self.critical_trigger
-        )
-        self.warning_trigger = self.create_alert_rule_trigger(
-            alert_rule=self.alert_rule, label="warning"
-        )
-        self.warning_trigger_action = self.create_alert_rule_trigger_action(
-            alert_rule_trigger=self.warning_trigger
-        )
-        _, _, _, self.detector, _, _, _, _ = migrate_alert_rule(self.alert_rule)
-        self.critical_detector_trigger, _ = migrate_metric_data_conditions(self.critical_trigger)
-        self.warning_detector_trigger, _ = migrate_metric_data_conditions(self.warning_trigger)
-        self.critical_action, _, _ = migrate_metric_action(self.critical_trigger_action)
-        self.warning_action, _, _ = migrate_metric_action(self.warning_trigger_action)
-        self.resolve_trigger_data_condition = migrate_resolve_threshold_data_condition(
-            self.alert_rule
-        )
-        self.expected_actions = [
-            {
-                "id": str(self.critical_trigger_action.id),
-                "alertRuleTriggerId": str(self.critical_trigger.id),
-                "type": "email",
-                "targetType": "user",
-                "targetIdentifier": str(self.user.id),
-                "inputChannelId": None,
-                "integrationId": None,
-                "sentryAppId": None,
-                "dateCreated": self.critical_trigger_action.date_added,
-                "desc": f"Send a notification to {self.user.email}",
-                "priority": self.critical_action.data.get("priority"),
-            },
-        ]
-        self.expected_trigger = {
-            "id": str(self.critical_trigger.id),
-            "alertRuleId": str(self.alert_rule.id),
-            "label": "critical",
-            "thresholdType": AlertRuleThresholdType.ABOVE.value,
-            "alertThreshold": self.critical_detector_trigger.comparison,
-            "resolveThreshold": AlertRuleThresholdType.BELOW,
-            "dateCreated": self.critical_trigger.date_added,
-            "actions": self.expected_actions,
-        }
+        super().setUp()
+        self.add_warning_trigger()
 
     def create_rule_triggers_and_actions(
         self,
@@ -102,29 +51,15 @@ def test_simple(self) -> None:
             self.user,
             WorkflowEngineDataConditionSerializer(),
         )
-        assert serialized_data_condition == self.expected_trigger
+        assert serialized_data_condition == self.expected_triggers[0]
 
     def test_warning_trigger(self) -> None:
         serialized_data_condition = serialize(
             self.warning_detector_trigger,
             self.user,
             WorkflowEngineDataConditionSerializer(),
         )
-        expected_actions = self.expected_actions.copy()
-        expected_actions[0]["id"] = str(self.warning_trigger_action.id)
-        expected_actions[0]["alertRuleTriggerId"] = str(self.warning_trigger.id)
-
-        expected_trigger = self.expected_trigger.copy()
-        expected_trigger["actions"] = expected_actions
-        expected_trigger["alertThreshold"] = translate_data_condition_type(
-            self.detector.config.get("comparison_delta"),
-            self.resolve_trigger_data_condition.type,
-            self.warning_detector_trigger.comparison,
-        )
-        expected_trigger["id"] = str(self.warning_trigger.id)
-        expected_trigger["alertRuleId"] = str(self.alert_rule.id)
-        expected_trigger["label"] = "warning"
-        assert serialized_data_condition == expected_trigger
+        assert serialized_data_condition == self.expected_triggers[1]
 
     def test_multiple_actions(self) -> None:
         self.critical_trigger_action_2 = self.create_alert_rule_trigger_action(
@@ -136,7 +71,7 @@ def test_multiple_actions(self) -> None:
             self.user,
             WorkflowEngineDataConditionSerializer(),
         )
-        expected_actions = self.expected_actions.copy()
+        expected_actions = self.expected_critical_action.copy()
         actions_2 = {
             "id": str(self.critical_trigger_action_2.id),
             "alertRuleTriggerId": str(self.critical_trigger.id),
@@ -151,7 +86,7 @@ def test_multiple_actions(self) -> None:
             "priority": self.critical_action.data.get("priority"),
         }
         expected_actions.append(actions_2)
-        expected_trigger = self.expected_trigger.copy()
+        expected_trigger = self.expected_triggers[0].copy()
         expected_trigger["actions"] = expected_actions
         assert serialized_data_condition == expected_trigger
 
@@ -173,11 +108,11 @@ def test_comparison_delta(self) -> None:
             self.user,
             WorkflowEngineDataConditionSerializer(),
         )
-        expected_actions = self.expected_actions.copy()
+        expected_actions = self.expected_critical_action.copy()
         expected_actions[0]["id"] = str(comparison_delta_trigger_action.id)
         expected_actions[0]["alertRuleTriggerId"] = str(comparison_delta_trigger.id)
 
-        expected_trigger = self.expected_trigger.copy()
+        expected_trigger = self.expected_triggers[0].copy()
         expected_trigger["actions"] = expected_actions
         expected_trigger["alertThreshold"] = translate_data_condition_type(
             detector.config.get("comparison_delta"),