[CORRUPTED] Synthetic Benchmark PR #103870 - Add single datasource enforcement config

src/sentry/monitors/validators.py

@@ -716,6 +716,7 @@ class MonitorIncidentDetectorValidator(BaseDetectorTypeValidator):
     data_source field (MonitorDataSourceValidator).
     """
 
+    enforce_single_datasource = True
     data_sources = MonitorDataSourceListField(child=MonitorDataSourceValidator(), required=False)
 
     def validate_enabled(self, value: bool) -> bool:

src/sentry/uptime/endpoints/validators.py

@@ -464,6 +464,7 @@ def create_source(self, validated_data: dict[str, Any]) -> UptimeSubscription:
 
 
 class UptimeDomainCheckFailureValidator(BaseDetectorTypeValidator):
+    enforce_single_datasource = True
     data_sources = serializers.ListField(child=UptimeMonitorDataSourceValidator(), required=False)
 
     def validate_config(self, config: dict[str, Any]) -> dict[str, Any]:

src/sentry/workflow_engine/endpoints/validators/base/detector.py

@@ -41,6 +41,12 @@ class DetectorQuota:
 
 
 class BaseDetectorTypeValidator(CamelSnakeSerializer):
+    enforce_single_datasource = False
+    """
+    Set to True in subclasses to enforce that only a single data source can be configured.
+    This prevents invalid configurations for detector types that don't support multiple data sources.
+    """
+
     name = serializers.CharField(
         required=True,
         max_length=200,
@@ -79,6 +85,16 @@ def data_sources(self) -> serializers.ListField:
     def data_conditions(self) -> BaseDataConditionValidator:
         raise NotImplementedError
 
+    def validate_data_sources(self, value: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """
+        Validate data sources, enforcing single data source if configured.
+        """
+        if self.enforce_single_datasource and len(value) > 1:
+            raise serializers.ValidationError(
+                "Only one data source is allowed for this detector type."
+            )
+        return value
+
     def get_quota(self) -> DetectorQuota:
         return DetectorQuota(has_exceeded=False, limit=-1, count=-1)
 

tests/sentry/monitors/test_validators.py

@@ -1195,6 +1195,33 @@ def test_detector_requires_data_source(self):
         assert not validator.is_valid()
         assert "dataSources" in validator.errors
 
+    def test_rejects_multiple_data_sources(self):
+        """Test that multiple data sources are rejected for cron monitors."""
+        # Create a condition group for testing
+        condition_group = DataConditionGroup.objects.create(
+            organization_id=self.organization.id,
+            logic_type=DataConditionGroup.Type.ANY,
+        )
+        data = self._get_valid_detector_data(
+            dataSources=[
+                {
+                    "name": "Test Monitor 1",
+                    "slug": "test-monitor-1",
+                    "config": self._get_base_config(),
+                },
+                {
+                    "name": "Test Monitor 2",
+                    "slug": "test-monitor-2",
+                    "config": self._get_base_config(),
+                },
+            ]
+        )
+        context = {**self.context, "condition_group": condition_group}
+        validator = self._create_validator(data, context=context)
+        assert not validator.is_valid()
+        assert "dataSources" in validator.errors
+        assert "Only one data source is allowed" in str(validator.errors["dataSources"])
+
     def test_create_detector_validates_data_source(self):
         condition_group = DataConditionGroup.objects.create(
             organization_id=self.organization.id,

tests/sentry/uptime/endpoints/test_validators.py

@@ -146,6 +146,27 @@ def get_valid_data(self, **kwargs):
             ),
         }
 
+    def test_rejects_multiple_data_sources(self):
+        """Test that multiple data sources are rejected for uptime monitors."""
+        data = self.get_valid_data(
+            data_sources=[
+                {
+                    "url": "https://sentry.io",
+                    "intervalSeconds": 60,
+                    "timeoutMs": 1000,
+                },
+                {
+                    "url": "https://example.com",
+                    "intervalSeconds": 60,
+                    "timeoutMs": 1000,
+                },
+            ]
+        )
+        validator = UptimeDomainCheckFailureValidator(data=data, context=self.context)
+        assert not validator.is_valid()
+        assert "dataSources" in validator.errors
+        assert "Only one data source is allowed" in str(validator.errors["dataSources"])
+
     @mock.patch(
         "sentry.quotas.backend.assign_seat",
         return_value=Outcome.ACCEPTED,