feat(crons): Add get_detector_for_monitor function

wedamija · wedamija · commit 1c4782e0dffd · 2025-08-08T16:42:23.000-07:00
Add a simple function to get the detector associated with a cron monitor, if one exists.
diff --git a/src/sentry/monitors/utils.py b/src/sentry/monitors/utils.py
@@ -410,3 +410,13 @@ def ensure_cron_detector(monitor: Monitor):
                 DataSourceDetector.objects.create(data_source=data_source, detector=detector)
     except Exception:
         logger.exception("Error creating cron detector")
+
+
+def get_detector_for_monitor(monitor: Monitor) -> Detector | None:
+    try:
+        return Detector.objects.get(
+            datasource__type=DATA_SOURCE_CRON_MONITOR,
+            datasource__source_id=str(monitor.id),
+        )
+    except Detector.DoesNotExist:
+        return None
diff --git a/tests/sentry/monitors/consumers/test_monitor_consumer.py b/tests/sentry/monitors/consumers/test_monitor_consumer.py
@@ -29,14 +29,14 @@
     ScheduleType,
 )
 from sentry.monitors.processing_errors.errors import ProcessingErrorsException, ProcessingErrorType
-from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR, CheckinItem
+from sentry.monitors.types import CheckinItem
+from sentry.monitors.utils import get_detector_for_monitor
 from sentry.testutils.asserts import assert_org_audit_log_exists
 from sentry.testutils.cases import TestCase
 from sentry.testutils.helpers.options import override_options
 from sentry.testutils.outbox import outbox_runner
 from sentry.utils import json
 from sentry.utils.outcomes import Outcome
-from sentry.workflow_engine.models import Detector
 
 
 class ExpectNoProcessingError:
@@ -573,10 +573,7 @@ def test_monitor_create(self) -> None:
             monitor_environment.next_checkin_latest
             == monitor_environment.monitor.get_next_expected_checkin_latest(checkin.date_added)
         )
-        assert Detector.objects.filter(
-            datasource__type=DATA_SOURCE_CRON_MONITOR,
-            datasource__source_id=str(monitor_environment.monitor_id),
-        ).exists()
+        get_detector_for_monitor(monitor_environment.monitor)
 
     def test_monitor_create_owner(self) -> None:
         self.send_checkin(
diff --git a/tests/sentry/monitors/endpoints/test_organization_monitor_index.py b/tests/sentry/monitors/endpoints/test_organization_monitor_index.py
@@ -12,14 +12,13 @@
 from sentry.constants import ObjectStatus
 from sentry.models.rule import Rule, RuleSource
 from sentry.monitors.models import Monitor, MonitorStatus, ScheduleType
-from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
+from sentry.monitors.utils import get_detector_for_monitor
 from sentry.quotas.base import SeatAssignmentResult
 from sentry.slug.errors import DEFAULT_SLUG_ERROR_MESSAGE
 from sentry.testutils.asserts import assert_org_audit_log_exists
 from sentry.testutils.cases import MonitorTestCase
 from sentry.testutils.outbox import outbox_runner
 from sentry.utils.outcomes import Outcome
-from sentry.workflow_engine.models import Detector
 
 
 class ListOrganizationMonitorsTest(MonitorTestCase):
@@ -404,11 +403,7 @@ def test_simple(self, mock_record: MagicMock) -> None:
             data={"upsert": False, **monitor.get_audit_log_data()},
         )
 
-        assert Detector.objects.filter(
-            datasource__type=DATA_SOURCE_CRON_MONITOR,
-            datasource__source_id=str(monitor.id),
-        ).exists()
-
+        assert get_detector_for_monitor(monitor) is not None
         self.project.refresh_from_db()
         assert self.project.flags.has_cron_monitors
 
diff --git a/tests/sentry/monitors/test_utils.py b/tests/sentry/monitors/test_utils.py
@@ -3,9 +3,9 @@
 from django.db import IntegrityError
 
 from sentry.monitors.types import DATA_SOURCE_CRON_MONITOR
-from sentry.monitors.utils import ensure_cron_detector
+from sentry.monitors.utils import ensure_cron_detector, get_detector_for_monitor
 from sentry.testutils.cases import TestCase
-from sentry.workflow_engine.models import DataSource, DataSourceDetector, Detector
+from sentry.workflow_engine.models import DataSource, Detector
 
 
 class EnsureCronDetectorTest(TestCase):
@@ -14,66 +14,22 @@ def setUp(self):
         self.monitor = self.create_monitor(owner_user_id=None)
 
     def test_creates_data_source_and_detector_for_new_monitor(self):
-        assert not DataSource.objects.filter(
-            type=DATA_SOURCE_CRON_MONITOR,
-            organization_id=self.monitor.organization_id,
-            source_id=str(self.monitor.id),
-        ).exists()
-
+        assert not get_detector_for_monitor(self.monitor)
         ensure_cron_detector(self.monitor)
-        data_source = DataSource.objects.get(
-            type=DATA_SOURCE_CRON_MONITOR,
-            organization_id=self.monitor.organization_id,
-            source_id=str(self.monitor.id),
-        )
-        assert data_source is not None
-        detector = Detector.objects.get(
-            type="monitor_check_in_failure",
-            project_id=self.monitor.project_id,
-            name=self.monitor.name,
-        )
+        detector = get_detector_for_monitor(self.monitor)
         assert detector is not None
+        assert detector.type == "monitor_check_in_failure"
+        assert detector.project_id == self.monitor.project_id
+        assert detector.name == self.monitor.name
         assert detector.owner_user_id == self.monitor.owner_user_id
         assert detector.owner_team_id == self.monitor.owner_team_id
-        assert DataSourceDetector.objects.filter(
-            data_source=data_source,
-            detector=detector,
-        ).exists()
 
     def test_idempotent_for_existing_data_source(self):
         ensure_cron_detector(self.monitor)
-        data_source = DataSource.objects.get(
-            type=DATA_SOURCE_CRON_MONITOR,
-            organization_id=self.monitor.organization_id,
-            source_id=str(self.monitor.id),
-        )
-        detector = Detector.objects.get(
-            type="monitor_check_in_failure",
-            project_id=self.monitor.project_id,
-            name=self.monitor.name,
-        )
-        link = DataSourceDetector.objects.get(
-            data_source=data_source,
-            detector=detector,
-        )
+        detector = get_detector_for_monitor(self.monitor)
         ensure_cron_detector(self.monitor)
-        data_source_after = DataSource.objects.get(
-            type=DATA_SOURCE_CRON_MONITOR,
-            organization_id=self.monitor.organization_id,
-            source_id=str(self.monitor.id),
-        )
-        detector_after = Detector.objects.get(
-            type="monitor_check_in_failure",
-            project_id=self.monitor.project_id,
-            name=self.monitor.name,
-        )
-        link_after = DataSourceDetector.objects.get(
-            data_source=data_source,
-            detector=detector,
-        )
-        assert data_source.id == data_source_after.id
+        detector_after = get_detector_for_monitor(self.monitor)
         assert detector.id == detector_after.id
-        assert link.id == link_after.id
 
     def test_with_owner_user(self):
         self.monitor.owner_user_id = self.user.id
@@ -117,3 +73,38 @@ def test_atomic_transaction_rollback(self):
         assert not DataSource.objects.filter(
             type=DATA_SOURCE_CRON_MONITOR, source_id=str(self.monitor.id)
         ).exists()
+
+
+class GetDetectorForMonitorTest(TestCase):
+    def setUp(self):
+        super().setUp()
+        self.monitor = self.create_monitor()
+
+    def test_returns_none_when_no_detector_exists(self):
+        detector = get_detector_for_monitor(self.monitor)
+        assert detector is None
+
+    def test_returns_detector_when_exists(self):
+        ensure_cron_detector(self.monitor)
+
+        detector = get_detector_for_monitor(self.monitor)
+        assert detector is not None
+        assert detector.type == "monitor_check_in_failure"
+        assert detector.project_id == self.monitor.project_id
+        assert detector.name == self.monitor.name
+
+    def test_returns_correct_detector_for_specific_monitor(self):
+        monitor1 = self.monitor
+        monitor2 = self.create_monitor(name="Monitor 2")
+
+        ensure_cron_detector(monitor1)
+        ensure_cron_detector(monitor2)
+
+        detector1 = get_detector_for_monitor(monitor1)
+        detector2 = get_detector_for_monitor(monitor2)
+
+        assert detector1 is not None
+        assert detector2 is not None
+        assert detector1.id != detector2.id
+        assert detector1.name == monitor1.name
+        assert detector2.name == monitor2.name
