feat(uptime): Add billing seat management for detector validators

Add billing seat assignment and removal to UptimeDomainCheckFailureValidator
to ensure uptime detectors created, updated, or deleted via the detector APIs
correctly handle billing seats.

Changes:
- Assign seats when creating enabled detectors (gracefully disable if no seats)
- Assign/remove seats when enabling/disabling detectors via updates
- Remove seats immediately when deleting detectors
- Raise validation errors when enabling fails due to no available seats

Also add comprehensive tests to verify all billing seat operations work
correctly across create, update, and delete operations.

Fixes [NEW-527: Ensure CRUD / enable+disable detectors in new UI handles assigning / unassigning seats](https://linear.app/getsentry/issue/NEW-527/ensure-crud-enabledisable-detectors-in-new-ui-handles-assigning)

Author: evanpurkhiser

src/sentry/uptime/endpoints/validators.py

@@ -7,11 +7,11 @@
 from rest_framework import serializers
 from rest_framework.fields import URLField
 
-from sentry import audit_log
+from sentry import audit_log, quotas
 from sentry.api.fields import ActorField
 from sentry.api.serializers.rest_framework import CamelSnakeSerializer
 from sentry.auth.superuser import is_active_superuser
-from sentry.constants import ObjectStatus
+from sentry.constants import DataCategory, ObjectStatus
 from sentry.models.environment import Environment
 from sentry.uptime.models import (
     UptimeSubscription,
@@ -28,6 +28,10 @@
     check_url_limits,
     create_uptime_detector,
     create_uptime_subscription,
+    delete_uptime_subscription,
+    disable_uptime_detector,
+    enable_uptime_detector,
+    remove_uptime_seat,
     update_uptime_detector,
     update_uptime_subscription,
 )
@@ -462,7 +466,47 @@ def create_source(self, validated_data: dict[str, Any]) -> UptimeSubscription:
 class UptimeDomainCheckFailureValidator(BaseDetectorTypeValidator):
     data_sources = serializers.ListField(child=UptimeMonitorDataSourceValidator(), required=False)
 
+    def validate_enabled(self, value: bool) -> bool:
+        """
+        Validate that enabling a detector is allowed based on seat availability.
+
+        This check will ONLY be performed when a detector instance is provided via
+        context (i.e., during updates). For creation, seat assignment is handled
+        in the create() method after the detector is created.
+        """
+        detector = self.instance
+
+        # Only validate on updates when trying to enable a currently disabled detector
+        if detector and value and not detector.enabled:
+            result = quotas.backend.check_assign_seat(DataCategory.UPTIME, detector)
+            if not result.assignable:
+                raise serializers.ValidationError(result.reason)
+
+        return value
+
+    def create(self, validated_data):
+        detector = super().create(validated_data)
+
+        try:
+            enable_uptime_detector(detector, ensure_assignment=True)
+        except UptimeMonitorNoSeatAvailable:
+            # No need to do anything if we failed to handle seat
+            # assignment. The monitor will be created, but not enabled
+            pass
+
+        return detector
+
     def update(self, instance: Detector, validated_data: dict[str, Any]) -> Detector:
+        # Handle seat management when enabling/disabling
+        was_enabled = instance.enabled
+        enabled = validated_data.get("enabled", was_enabled)
+
+        if was_enabled != enabled:
+            if enabled:
+                enable_uptime_detector(instance)
+            else:
+                disable_uptime_detector(instance)
+
         super().update(instance, validated_data)
 
         data_source = None
@@ -497,5 +541,11 @@ def update_data_source(self, instance: Detector, data_source: dict[str, Any]):
 
         return instance
 
-    def create(self, validated_data):
-        return super().create(validated_data)
+    def delete(self) -> None:
+        assert self.instance is not None
+
+        remove_uptime_seat(self.instance)
+        uptime_subscription = get_uptime_subscription(self.instance)
+        delete_uptime_subscription(uptime_subscription)
+
+        super().delete()

src/sentry/uptime/subscriptions/subscriptions.py

@@ -476,6 +476,18 @@ def disable_uptime_detector(detector: Detector, skip_quotas: bool = False):
             delete_remote_uptime_subscription.delay(uptime_subscription.id)
 
 
+def ensure_uptime_seat(detector: Detector) -> None:
+    """
+    Ensures that a billing seat is assigned for the uptime detector.
+
+    Raises UptimeMonitorNoSeatAvailable if no seats are available.
+    """
+    outcome = quotas.backend.assign_seat(DataCategory.UPTIME, detector)
+    if outcome != Outcome.ACCEPTED:
+        result = quotas.backend.check_assign_seat(DataCategory.UPTIME, detector)
+        raise UptimeMonitorNoSeatAvailable(result)
+
+
 def enable_uptime_detector(
     detector: Detector, ensure_assignment: bool = False, skip_quotas: bool = False
 ):
@@ -494,11 +506,11 @@ def enable_uptime_detector(
         return
 
     if not skip_quotas:
-        outcome = quotas.backend.assign_seat(DataCategory.UPTIME, detector)
-        if outcome != Outcome.ACCEPTED:
-            disable_uptime_detector(detector)
-            result = quotas.backend.check_assign_seat(DataCategory.UPTIME, detector)
-            raise UptimeMonitorNoSeatAvailable(result)
+        try:
+            ensure_uptime_seat(detector)
+        except UptimeMonitorNoSeatAvailable:
+            disable_uptime_detector(detector, skip_quotas=True)
+            raise
 
     uptime_subscription: UptimeSubscription = get_uptime_subscription(detector)
     detector.update(enabled=True)
@@ -513,9 +525,14 @@ def enable_uptime_detector(
         )
 
 
+def remove_uptime_seat(detector: Detector):
+    quotas.backend.remove_seat(DataCategory.UPTIME, detector)
+
+
 def delete_uptime_detector(detector: Detector):
     uptime_subscription = get_uptime_subscription(detector)
-    quotas.backend.remove_seat(DataCategory.UPTIME, detector)
+
+    remove_uptime_seat(detector)
     detector.update(status=ObjectStatus.PENDING_DELETION)
     RegionScheduledDeletion.schedule(detector, days=0)
     delete_uptime_subscription(uptime_subscription)

tests/sentry/uptime/endpoints/test_validators.py

@@ -1,9 +1,21 @@
+from unittest import mock
+
+from sentry.constants import DataCategory
+from sentry.quotas.base import SeatAssignmentResult
 from sentry.testutils.cases import TestCase, UptimeTestCase
 from sentry.uptime.endpoints.validators import (
+    UptimeDomainCheckFailureValidator,
     UptimeMonitorDataSourceValidator,
     compute_http_request_size,
 )
-from sentry.uptime.models import UptimeSubscription
+from sentry.uptime.grouptype import UptimeDomainCheckFailure
+from sentry.uptime.models import UptimeSubscription, get_uptime_subscription
+from sentry.uptime.types import (
+    DEFAULT_DOWNTIME_THRESHOLD,
+    DEFAULT_RECOVERY_THRESHOLD,
+    UptimeMonitorMode,
+)
+from sentry.utils.outcomes import Outcome
 
 
 class ComputeHttpRequestSizeTest(UptimeTestCase):
@@ -97,3 +109,188 @@ def test_too_big_request(self):
         data = self.get_valid_data(body="0" * 1000)
         validator = UptimeMonitorDataSourceValidator(data=data, context=self.context)
         assert not validator.is_valid()
+
+
+class UptimeDomainCheckFailureValidatorTest(UptimeTestCase):
+    def setUp(self):
+        super().setUp()
+        self.context = {
+            "organization": self.organization,
+            "project": self.project,
+            "request": self.make_request(user=self.user),
+        }
+
+    def get_valid_data(self, **kwargs):
+        return {
+            "name": kwargs.get("name", "Test Uptime Monitor"),
+            "type": UptimeDomainCheckFailure.slug,
+            "enabled": kwargs.get("enabled", True),
+            "config": kwargs.get(
+                "config",
+                {
+                    "mode": UptimeMonitorMode.MANUAL.value,
+                    "environment": None,
+                    "recovery_threshold": DEFAULT_RECOVERY_THRESHOLD,
+                    "downtime_threshold": DEFAULT_DOWNTIME_THRESHOLD,
+                },
+            ),
+            "dataSources": kwargs.get(
+                "data_sources",
+                [
+                    {
+                        "url": "https://sentry.io",
+                        "intervalSeconds": 60,
+                        "timeoutMs": 1000,
+                    }
+                ],
+            ),
+        }
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.ACCEPTED,
+    )
+    def test_create_enabled_assigns_seat(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that creating an enabled detector assigns a billing seat."""
+        validator = UptimeDomainCheckFailureValidator(
+            data=self.get_valid_data(enabled=True), context=self.context
+        )
+        assert validator.is_valid(), validator.errors
+        detector = validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is True
+
+        # Verify seat was assigned
+        mock_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.RATE_LIMITED,
+    )
+    def test_create_enabled_no_seat_available(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that creating a detector with no seats available creates it but leaves it disabled."""
+        validator = UptimeDomainCheckFailureValidator(
+            data=self.get_valid_data(enabled=True), context=self.context
+        )
+        assert validator.is_valid(), validator.errors
+        detector = validator.save()
+
+        detector.refresh_from_db()
+        # Detector created but not enabled due to no seat assignment
+        assert detector.enabled is False
+
+        # Verify seat assignment was attempted
+        mock_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+        uptime_subscription = get_uptime_subscription(detector)
+        assert uptime_subscription.status == UptimeSubscription.Status.DISABLED.value
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.ACCEPTED,
+    )
+    def test_update_enable_assigns_seat(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that enabling a previously disabled detector assigns a seat."""
+        # Create a disabled detector
+        detector = self.create_uptime_detector(enabled=False)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"enabled": True}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is True
+
+        # Verify seat was assigned
+        mock_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+        uptime_subscription = get_uptime_subscription(detector)
+        assert uptime_subscription.status == UptimeSubscription.Status.ACTIVE.value
+
+    @mock.patch(
+        "sentry.quotas.backend.check_assign_seat",
+        return_value=SeatAssignmentResult(assignable=False, reason="No seats available"),
+    )
+    def test_update_enable_no_seat_available(self, mock_check_assign_seat: mock.MagicMock) -> None:
+        """Test that enabling fails with validation error when no seats are available."""
+        # Create a disabled detector
+        detector = self.create_uptime_detector(enabled=False)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"enabled": True}, context=self.context, partial=True
+        )
+
+        # Validation should fail due to no seats available
+        assert not validator.is_valid()
+        assert "enabled" in validator.errors
+        assert validator.errors["enabled"] == ["No seats available"]
+
+        detector.refresh_from_db()
+        # Detector should still be disabled
+        assert detector.enabled is False
+
+        # Verify seat availability check was performed
+        mock_check_assign_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+    @mock.patch("sentry.quotas.backend.disable_seat")
+    def test_update_disable_removes_seat(self, mock_disable_seat: mock.MagicMock) -> None:
+        """Test that disabling a previously enabled detector removes the seat."""
+        # Create an enabled detector
+        detector = self.create_uptime_detector(enabled=True)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"enabled": False}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        assert detector.enabled is False
+
+        # Verify disable_seat was called
+        mock_disable_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+        uptime_subscription = get_uptime_subscription(detector)
+        assert uptime_subscription.status == UptimeSubscription.Status.DISABLED.value
+
+    @mock.patch("sentry.quotas.backend.remove_seat")
+    def test_delete_removes_seat(self, mock_remove_seat: mock.MagicMock) -> None:
+        """Test that deleting a detector removes its billing seat immediately."""
+        detector = self.create_uptime_detector(enabled=True)
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={}, context=self.context
+        )
+
+        validator.delete()
+
+        # Verify remove_seat was called immediately
+        mock_remove_seat.assert_called_with(DataCategory.UPTIME, detector)
+
+    @mock.patch(
+        "sentry.quotas.backend.assign_seat",
+        return_value=Outcome.ACCEPTED,
+    )
+    def test_update_no_enable_change_no_seat_call(self, mock_assign_seat: mock.MagicMock) -> None:
+        """Test that updating without changing enabled status doesn't trigger seat operations."""
+        # Create an enabled detector
+        detector = self.create_uptime_detector(enabled=True)
+
+        # Clear any previous mock calls from creation
+        mock_assign_seat.reset_mock()
+
+        validator = UptimeDomainCheckFailureValidator(
+            instance=detector, data={"name": "Updated Name"}, context=self.context, partial=True
+        )
+        assert validator.is_valid(), validator.errors
+        validator.save()
+
+        detector.refresh_from_db()
+        assert detector.name == "Updated Name"
+        assert detector.enabled is True
+
+        # Verify no seat operations were called
+        mock_assign_seat.assert_not_called()