chore(ACI): Add back subscription processor test coverage (#105113)

ceorourke · web-flow · commit 1817fb71aff2 · 2025-12-19T09:41:00.000-08:00
Add missing test coverage for the subscription processor. Many tests we still need were deleted and need to be recovered but updated for ACI. * Add missing tests for early returns from [here](https://github.com/getsentry/sentry/blob/e3f3190fcfcc4ff63e3562516ab92f641d4d1680/tests/sentry/incidents/subscription_processor/test_subscription_processor.py#L287-L388) * Add missing anomaly detection test coverage from [here](https://github.com/getsentry/sentry/blob/e3f3190fcfcc4ff63e3562516ab92f641d4d1680/tests/sentry/incidents/subscription_processor/test_subscription_processor.py#L3138) TODO in future PRs: - [x] Add missing crash rate alert tests from [here](https://github.com/getsentry/sentry/blob/e3f3190fcfcc4ff63e3562516ab92f641d4d1680/tests/sentry/incidents/subscription_processor/test_subscription_processor.py#L3741) #105205 - [x] Add missing upsample tests from [here](https://github.com/getsentry/sentry/blob/e3f3190fcfcc4ff63e3562516ab92f641d4d1680/tests/sentry/incidents/subscription_processor/test_subscription_processor.py#L4626) #105272 * Add missing EAP test (I can't get this one to pass, it always receives `None` from `get_eap_aggregation_value` so I will ask Shruthi post holidays) from [here](https://github.com/getsentry/sentry/blob/e3f3190fcfcc4ff63e3562516ab92f641d4d1680/tests/sentry/incidents/subscription_processor/test_subscription_processor.py#L1869)
diff --git a/tests/sentry/incidents/handlers/condition/test_anomaly_detection_handler.py b/tests/sentry/incidents/handlers/condition/test_anomaly_detection_handler.py
@@ -1,9 +1,11 @@
 from datetime import UTC, datetime
+from typing import Any
 from unittest import mock
 
 import orjson
 from urllib3.response import HTTPResponse
 
+from sentry.conf.server import SEER_ANOMALY_DETECTION_ENDPOINT_URL
 from sentry.incidents.utils.types import AnomalyDetectionUpdate
 from sentry.seer.anomaly_detection.types import (
     AnomalyDetectionSeasonality,
@@ -13,6 +15,7 @@
     DataSourceType,
     DetectAnomaliesResponse,
 )
+from sentry.snuba.dataset import Dataset
 from sentry.snuba.subscriptions import create_snuba_subscription
 from sentry.workflow_engine.models import Condition, DataPacket
 from sentry.workflow_engine.types import ConditionError, DetectorPriorityLevel
@@ -44,7 +47,6 @@ def setUp(self) -> None:
             timestamp=datetime.now(UTC),
             entity="test-entity",
         )
-
         self.data_source = self.create_data_source(
             source_id=str(packet.subscription_id),
             organization=self.organization,
@@ -75,43 +77,106 @@ def setUp(self) -> None:
                         "anomaly_type": AnomalyType.HIGH_CONFIDENCE,
                     },
                     "timestamp": 1,
-                    "value": 10,
+                    "value": self.data_packet.packet.values["value"],
                 }
             ],
         }
+        self.low_confidence_seer_response: DetectAnomaliesResponse = {
+            "success": True,
+            "timeseries": [
+                {
+                    "anomaly": {
+                        "anomaly_score": 0.2,
+                        "anomaly_type": AnomalyType.LOW_CONFIDENCE,
+                    },
+                    "timestamp": 1,
+                    "value": self.data_packet.packet.values["value"],
+                }
+            ],
+        }
+
+    def assert_seer_call(self, deserialized_body: dict[str, Any]) -> None:
+        assert deserialized_body["organization_id"] == self.detector.project.organization.id
+        assert deserialized_body["project_id"] == self.detector.project_id
+        assert deserialized_body["config"]["time_period"] == self.snuba_query.time_window / 60
+        assert (
+            deserialized_body["config"]["sensitivity"]
+            == self.dc.comparison.get("sensitivity").value
+        )
+        assert (
+            deserialized_body["config"]["expected_seasonality"]
+            == self.dc.comparison.get("seasonality").value
+        )
+        assert deserialized_body["context"]["source_id"] == self.subscription.id
+        assert (
+            deserialized_body["context"]["source_type"] == DataSourceType.SNUBA_QUERY_SUBSCRIPTION
+        )
+        assert (
+            deserialized_body["context"]["cur_window"]["value"]
+            == self.data_packet.packet.values["value"]
+        )
 
     @mock.patch(
         "sentry.seer.anomaly_detection.get_anomaly_data.SEER_ANOMALY_DETECTION_CONNECTION_POOL.urlopen"
     )
-    def test_passes(self, mock_seer_request: mock.MagicMock) -> None:
-        seer_return_value = self.high_confidence_seer_response
-        mock_seer_request.return_value = HTTPResponse(orjson.dumps(seer_return_value), status=200)
+    def test_triggers(self, mock_seer_request: mock.MagicMock) -> None:
+        mock_seer_request.return_value = HTTPResponse(
+            orjson.dumps(self.high_confidence_seer_response), status=200
+        )
         assert (
             self.dc.evaluate_value(self.data_packet.packet.values)
             == DetectorPriorityLevel.HIGH.value
         )
+        assert mock_seer_request.call_args.args[0] == "POST"
+        assert mock_seer_request.call_args.args[1] == SEER_ANOMALY_DETECTION_ENDPOINT_URL
+        deserialized_body = orjson.loads(mock_seer_request.call_args.kwargs["body"])
+        self.assert_seer_call(deserialized_body)
 
     @mock.patch(
         "sentry.seer.anomaly_detection.get_anomaly_data.SEER_ANOMALY_DETECTION_CONNECTION_POOL.urlopen"
     )
-    def test_passes_medium(self, mock_seer_request: mock.MagicMock) -> None:
-        seer_return_value: DetectAnomaliesResponse = {
-            "success": True,
-            "timeseries": [
-                {
-                    "anomaly": {
-                        "anomaly_score": 0.2,
-                        "anomaly_type": AnomalyType.LOW_CONFIDENCE,
-                    },
-                    "timestamp": 1,
-                    "value": 10,
-                }
-            ],
-        }
-        mock_seer_request.return_value = HTTPResponse(orjson.dumps(seer_return_value), status=200)
+    def test_does_not_trigger(self, mock_seer_request: mock.MagicMock) -> None:
+        mock_seer_request.return_value = HTTPResponse(
+            orjson.dumps(self.low_confidence_seer_response), status=200
+        )
+        assert (
+            self.dc.evaluate_value(self.data_packet.packet.values) == DetectorPriorityLevel.OK.value
+        )
+        assert mock_seer_request.call_args.args[0] == "POST"
+        assert mock_seer_request.call_args.args[1] == SEER_ANOMALY_DETECTION_ENDPOINT_URL
+        deserialized_body = orjson.loads(mock_seer_request.call_args.kwargs["body"])
+        self.assert_seer_call(deserialized_body)
+
+    @mock.patch(
+        "sentry.seer.anomaly_detection.get_anomaly_data.SEER_ANOMALY_DETECTION_CONNECTION_POOL.urlopen"
+    )
+    def test_triggers_performance_detector(self, mock_seer_request: mock.MagicMock) -> None:
+        self.snuba_query.update(time_window=15 * 60, dataset=Dataset.Transactions)
+
+        # ensure that it triggers
+        mock_seer_request.return_value = HTTPResponse(
+            orjson.dumps(self.high_confidence_seer_response), status=200
+        )
+        assert (
+            self.dc.evaluate_value(self.data_packet.packet.values)
+            == DetectorPriorityLevel.HIGH.value
+        )
+        assert mock_seer_request.call_args.args[0] == "POST"
+        assert mock_seer_request.call_args.args[1] == SEER_ANOMALY_DETECTION_ENDPOINT_URL
+        deserialized_body = orjson.loads(mock_seer_request.call_args.kwargs["body"])
+        self.assert_seer_call(deserialized_body)
+
+        # ensure that it resolves
+        mock_seer_request.return_value = HTTPResponse(
+            orjson.dumps(self.low_confidence_seer_response), status=200
+        )
         assert (
             self.dc.evaluate_value(self.data_packet.packet.values) == DetectorPriorityLevel.OK.value
         )
+        assert mock_seer_request.call_args.args[0] == "POST"
+        assert mock_seer_request.call_args.args[1] == SEER_ANOMALY_DETECTION_ENDPOINT_URL
+        deserialized_body = orjson.loads(mock_seer_request.call_args.kwargs["body"])
+        self.assert_seer_call(deserialized_body)
 
     @mock.patch(
         "sentry.seer.anomaly_detection.get_anomaly_data.SEER_ANOMALY_DETECTION_CONNECTION_POOL.urlopen"
diff --git a/tests/sentry/incidents/subscription_processor/test_subscription_processor_aci.py b/tests/sentry/incidents/subscription_processor/test_subscription_processor_aci.py
@@ -1,10 +1,13 @@
 import copy
 from datetime import timedelta
 from functools import cached_property
-from unittest.mock import call, patch
+from unittest.mock import MagicMock, call, patch
 
 from django.utils import timezone
 
+from sentry.constants import ObjectStatus
+from sentry.incidents.subscription_processor import SubscriptionProcessor
+from sentry.snuba.dataset import Dataset
 from sentry.testutils.factories import DEFAULT_EVENT_DATA
 from sentry.workflow_engine.models.data_condition import Condition, DataCondition
 from sentry.workflow_engine.types import DetectorPriorityLevel
@@ -18,80 +21,158 @@ class ProcessUpdateTest(ProcessUpdateBaseClass):
     Test early return scenarios + simple cases.
     """
 
-    # TODO: tests for early return scenarios. These will need to be added once
-    # we've decoupled the subscription processor from the alert rule model.
-
     def test_simple(self) -> None:
         """
         Verify that an alert can trigger.
         """
         self.send_update(self.critical_threshold + 1)
         assert self.get_detector_state(self.metric_detector) == DetectorPriorityLevel.HIGH
 
+    @patch("sentry.incidents.subscription_processor.metrics")
+    def test_missing_project(self, mock_metrics: MagicMock) -> None:
+        metrics_call = "incidents.alert_rules.ignore_deleted_project"
+
+        self.sub.project.update(status=ObjectStatus.PENDING_DELETION)
+        self.sub.project.save()
+
+        assert self.send_update(self.critical_threshold + 1) is False
+        mock_metrics.incr.assert_has_calls([call(metrics_call)])
+
+        mock_metrics.reset_mock()
+
+        self.sub.project.delete()
+        assert self.send_update(self.critical_threshold + 1) is False
+        mock_metrics.incr.assert_has_calls([call(metrics_call)])
+
+    @patch("sentry.incidents.subscription_processor.metrics")
+    def test_has_downgraded_incidents(self, mock_metrics: MagicMock) -> None:
+        processor = SubscriptionProcessor(self.sub)
+        message = self.build_subscription_update(
+            self.sub, value=self.critical_threshold + 1, time_delta=timedelta()
+        )
+
+        with self.capture_on_commit_callbacks(execute=True):
+            assert processor.process_update(message) is False
+            mock_metrics.incr.assert_has_calls(
+                [call("incidents.alert_rules.ignore_update_missing_incidents")]
+            )
+
+    @patch("sentry.incidents.subscription_processor.metrics")
+    def test_has_downgraded_incidents_performance(self, mock_metrics: MagicMock) -> None:
+        snuba_query = self.get_snuba_query(self.detector)
+        snuba_query.update(time_window=15 * 60, dataset=Dataset.Transactions.value)
+        snuba_query.save()
+
+        processor = SubscriptionProcessor(self.sub)
+        message = self.build_subscription_update(
+            self.sub, value=self.critical_threshold + 1, time_delta=timedelta()
+        )
+
+        with self.capture_on_commit_callbacks(execute=True):
+            assert processor.process_update(message) is False
+            mock_metrics.incr.assert_has_calls(
+                [call("incidents.alert_rules.ignore_update_missing_incidents_performance")]
+            )
+
+    @patch("sentry.incidents.subscription_processor.metrics")
+    def test_has_downgraded_on_demand(self, mock_metrics: MagicMock) -> None:
+        snuba_query = self.get_snuba_query(self.detector)
+        snuba_query.update(time_window=15 * 60, dataset=Dataset.PerformanceMetrics.value)
+        snuba_query.save()
+
+        processor = SubscriptionProcessor(self.sub)
+        message = self.build_subscription_update(
+            self.sub, value=self.critical_threshold + 1, time_delta=timedelta()
+        )
+
+        with self.capture_on_commit_callbacks(execute=True):
+            assert processor.process_update(message) is False
+            mock_metrics.incr.assert_has_calls(
+                [call("incidents.alert_rules.ignore_update_missing_on_demand")]
+            )
+
+    @patch("sentry.incidents.subscription_processor.metrics")
+    def test_skip_already_processed_update(self, mock_metrics: MagicMock) -> None:
+        assert self.send_update(value=self.critical_threshold + 1) is True
+        mock_metrics.incr.reset_mock()
+        assert self.send_update(value=self.critical_threshold + 1) is False
+
+        mock_metrics.incr.assert_called_once_with(
+            "incidents.alert_rules.skipping_already_processed_update"
+        )
+        mock_metrics.incr.reset_mock()
+        assert (
+            self.send_update(value=self.critical_threshold + 1, time_delta=timedelta(hours=1))
+            is True
+        )
+
+        mock_metrics.incr.assert_called_once_with("incidents.alert_rules.process_update.start")
+
     def test_resolve(self) -> None:
-        detector = self.metric_detector
         self.send_update(self.critical_threshold + 1, timedelta(minutes=-2))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.HIGH
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.HIGH
 
         self.send_update(self.resolve_threshold - 1, timedelta(minutes=-1))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.OK
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.OK
 
     def test_resolve_percent_boundary(self) -> None:
-        detector = self.metric_detector
-        self.update_threshold(detector, DetectorPriorityLevel.HIGH, 0.5)
-        self.update_threshold(detector, DetectorPriorityLevel.OK, 0.5)
+        self.update_threshold(self.detector, DetectorPriorityLevel.HIGH, 0.5)
+        self.update_threshold(self.detector, DetectorPriorityLevel.OK, 0.5)
         self.send_update(self.critical_threshold + 0.1, timedelta(minutes=-2))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.HIGH
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.HIGH
 
         self.send_update(self.resolve_threshold, timedelta(minutes=-1))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.OK
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.OK
 
     def test_reversed(self) -> None:
         """
         Test that resolutions work when the threshold is reversed.
         """
-        detector = self.metric_detector
-        DataCondition.objects.filter(condition_group=detector.workflow_condition_group).delete()
-        self.set_up_data_conditions(detector, Condition.LESS, 100, None, 100)
+        DataCondition.objects.filter(
+            condition_group=self.detector.workflow_condition_group
+        ).delete()
+        self.set_up_data_conditions(self.detector, Condition.LESS, 100, None, 100)
         self.send_update(self.critical_threshold - 1, timedelta(minutes=-2))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.HIGH
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.HIGH
 
         self.send_update(self.resolve_threshold, timedelta(minutes=-1))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.OK
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.OK
 
     def test_multiple_triggers(self) -> None:
-        detector = self.metric_detector
-        DataCondition.objects.filter(condition_group=detector.workflow_condition_group).delete()
-        self.set_up_data_conditions(detector, Condition.GREATER, 100, 50, 50)
+        DataCondition.objects.filter(
+            condition_group=self.detector.workflow_condition_group
+        ).delete()
+        self.set_up_data_conditions(self.detector, Condition.GREATER, 100, 50, 50)
 
         self.send_update(self.warning_threshold + 1, timedelta(minutes=-5))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.MEDIUM
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.MEDIUM
 
         self.send_update(self.critical_threshold + 1, timedelta(minutes=-4))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.HIGH
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.HIGH
 
         self.send_update(self.critical_threshold - 1, timedelta(minutes=-3))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.MEDIUM
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.MEDIUM
 
         self.send_update(self.warning_threshold - 1, timedelta(minutes=-2))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.OK
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.OK
 
     def test_multiple_triggers_reversed(self) -> None:
-        detector = self.metric_detector
-        DataCondition.objects.filter(condition_group=detector.workflow_condition_group).delete()
-        self.set_up_data_conditions(detector, Condition.LESS, 50, 100, 100)
+        DataCondition.objects.filter(
+            condition_group=self.detector.workflow_condition_group
+        ).delete()
+        self.set_up_data_conditions(self.detector, Condition.LESS, 50, 100, 100)
 
         self.send_update(self.warning_threshold - 1, timedelta(minutes=-5))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.MEDIUM
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.MEDIUM
 
         self.send_update(self.critical_threshold - 1, timedelta(minutes=-4))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.HIGH
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.HIGH
 
         self.send_update(self.critical_threshold + 1, timedelta(minutes=-3))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.MEDIUM
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.MEDIUM
 
         self.send_update(self.warning_threshold + 1, timedelta(minutes=-2))
-        assert self.get_detector_state(detector) == DetectorPriorityLevel.OK
+        assert self.get_detector_state(self.detector) == DetectorPriorityLevel.OK
 
     # TODO: the subscription processor has a 10 minute cooldown period for creating new incidents
     # We probably need similar logic within workflow engine.
@@ -100,25 +181,25 @@ def test_multiple_triggers_reversed(self) -> None:
 class ProcessUpdateComparisonAlertTest(ProcessUpdateBaseClass):
     @cached_property
     def comparison_detector_above(self):
-        detector = self.metric_detector
-        detector.config.update({"comparison_delta": 60 * 60})
-        detector.save()
-        self.update_threshold(detector, DetectorPriorityLevel.HIGH, 150)
-        self.update_threshold(detector, DetectorPriorityLevel.OK, 150)
-        snuba_query = self.get_snuba_query(detector)
-        snuba_query.update(time_window=60 * 60)
-        return detector
+        self.detector.config.update({"comparison_delta": 60 * 60})
+        self.detector.save()
+        self.update_threshold(self.detector, DetectorPriorityLevel.HIGH, 150)
+        self.update_threshold(self.detector, DetectorPriorityLevel.OK, 150)
+        self.snuba_query = self.get_snuba_query(self.detector)
+        self.snuba_query.update(time_window=60 * 60)
+        return self.detector
 
     @cached_property
     def comparison_detector_below(self):
-        detector = self.metric_detector
-        detector.config.update({"comparison_delta": 60 * 60})
-        detector.save()
-        DataCondition.objects.filter(condition_group=detector.workflow_condition_group).delete()
-        self.set_up_data_conditions(detector, Condition.LESS, 50, None, 50)
-        snuba_query = self.get_snuba_query(detector)
+        self.detector.config.update({"comparison_delta": 60 * 60})
+        self.detector.save()
+        DataCondition.objects.filter(
+            condition_group=self.detector.workflow_condition_group
+        ).delete()
+        self.set_up_data_conditions(self.detector, Condition.LESS, 50, None, 50)
+        snuba_query = self.get_snuba_query(self.detector)
         snuba_query.update(time_window=60 * 60)
-        return detector
+        return self.detector
 
     @patch("sentry.incidents.utils.process_update_helpers.metrics")
     def test_comparison_alert_above(self, helper_metrics):
diff --git a/tests/sentry/incidents/subscription_processor/test_subscription_processor_base.py b/tests/sentry/incidents/subscription_processor/test_subscription_processor_base.py
