feat(workflow_engine): Add DELETE Workflow Endpoint (#86246)

## Description
PR to add support for `DELETE` method on workflows. This will schedule
the deletion and use the safe workflow delete to ensure all subsequent
models are also cleaned up.

Author: saponifi3d

src/sentry/audit_log/register.py

@@ -589,23 +589,23 @@
 default_manager.add(
     AuditLogEvent(
         event_id=213,
-        name="WORFKLOW_ADD",
+        name="WORKFLOW_ADD",
         api_name="workflow.add",
         template="added workflow {name}",
     )
 )
 default_manager.add(
     AuditLogEvent(
         event_id=214,
-        name="WORFKLOW_EDIT",
+        name="WORKFLOW_EDIT",
         api_name="workflow.edit",
         template="edited workflow {name}",
     )
 )
 default_manager.add(
     AuditLogEvent(
         event_id=215,
-        name="WORFKLOW_REMOVE",
+        name="WORKFLOW_REMOVE",
         api_name="workflow.remove",
         template="removed workflow {name}",
     )

src/sentry/workflow_engine/endpoints/organization_workflow_details.py

@@ -2,6 +2,7 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
+from sentry import audit_log
 from sentry.api.api_owners import ApiOwner
 from sentry.api.api_publish_status import ApiPublishStatus
 from sentry.api.base import region_silo_endpoint
@@ -15,7 +16,9 @@
     RESPONSE_UNAUTHORIZED,
 )
 from sentry.apidocs.parameters import GlobalParams, WorkflowParams
+from sentry.deletions.models.scheduleddeletion import RegionScheduledDeletion
 from sentry.models.organization import Organization
+from sentry.utils.audit import create_audit_entry
 from sentry.workflow_engine.endpoints.serializers import WorkflowSerializer
 from sentry.workflow_engine.models import Workflow
 
@@ -69,10 +72,27 @@ def put(self, request: Request, organization: Organization, workflow: Workflow):
         """
         Updates a workflow
         """
+        create_audit_entry(
+            request=request,
+            organization=organization,
+            target_object=workflow.id,
+            event=audit_log.get_event_id("WORKFLOW_EDIT"),
+            data=workflow.get_audit_log_data(),
+        )
+
         pass
 
     def delete(self, request: Request, organization: Organization, workflow: Workflow):
         """
         Delete a workflow
         """
-        pass
+        RegionScheduledDeletion.schedule(workflow, days=0, actor=request.user)
+        create_audit_entry(
+            request=request,
+            organization=organization,
+            target_object=workflow.id,
+            event=audit_log.get_event_id("WORKFLOW_REMOVE"),
+            data=workflow.get_audit_log_data(),
+        )
+
+        return Response(status=204)

tests/sentry/workflow_engine/endpoints/test_organization_workflow_details.py

@@ -1,6 +1,15 @@
+from sentry import audit_log
 from sentry.api.serializers import serialize
+from sentry.deletions.models.scheduleddeletion import RegionScheduledDeletion
+from sentry.deletions.tasks.scheduled import run_scheduled_deletions
+from sentry.models.auditlogentry import AuditLogEntry
+from sentry.silo.base import SiloMode
 from sentry.testutils.cases import APITestCase
-from sentry.testutils.silo import region_silo_test
+from sentry.testutils.helpers import TaskRunner
+from sentry.testutils.outbox import outbox_runner
+from sentry.testutils.silo import assume_test_silo_mode, region_silo_test
+from sentry.workflow_engine.models import Action, DataConditionGroup
+from tests.sentry.workflow_engine.test_base import BaseWorkflowTest
 
 
 class OrganizationWorkflowDetailsBaseTest(APITestCase):
@@ -20,3 +29,92 @@ def test_simple(self):
 
     def test_does_not_exist(self):
         self.get_error_response(self.organization.slug, 3, status_code=404)
+
+
+@region_silo_test
+class OrganizationDeleteWorkflowTest(OrganizationWorkflowDetailsBaseTest, BaseWorkflowTest):
+    method = "DELETE"
+
+    def tasks(self):
+        return TaskRunner()
+
+    def setUp(self):
+        super().setUp()
+        self.workflow = self.create_workflow(organization_id=self.organization.id)
+
+    def test_simple(self):
+        with outbox_runner():
+            self.get_success_response(self.organization.slug, self.workflow.id)
+
+        assert RegionScheduledDeletion.objects.filter(
+            model_name="Workflow",
+            object_id=self.workflow.id,
+        ).exists()
+
+    def test_audit_entry(self):
+        with outbox_runner():
+            self.get_success_response(self.organization.slug, self.workflow.id)
+
+        with assume_test_silo_mode(SiloMode.CONTROL):
+            assert AuditLogEntry.objects.filter(
+                target_object=self.workflow.id,
+                event=audit_log.get_event_id("WORKFLOW_REMOVE"),
+                actor=self.user,
+            ).exists()
+
+    def test_does_not_exist(self):
+        with outbox_runner():
+            response = self.get_error_response(self.organization.slug, -1)
+            assert response.status_code == 404
+
+        # Ensure it wasn't deleted
+        assert not RegionScheduledDeletion.objects.filter(
+            model_name="Workflow",
+            object_id=self.workflow.id,
+        ).exists()
+
+    def test_delete_configured_workflow__action(self):
+        action_condition_group, action = self.create_workflow_action(workflow=self.workflow)
+
+        with outbox_runner():
+            self.get_success_response(self.organization.slug, self.workflow.id)
+
+        # Ensure the workflow is scheduled for deletion
+        assert RegionScheduledDeletion.objects.filter(
+            model_name="Workflow",
+            object_id=self.workflow.id,
+        ).exists()
+
+        # Delete the workflow
+        with self.tasks():
+            run_scheduled_deletions()
+
+        # Ensure action is removed
+        assert not Action.objects.filter(id=action.id).exists()
+
+    def test_delete_configured_workflow__action_condition(self):
+        action_condition_group, action = self.create_workflow_action(workflow=self.workflow)
+
+        with outbox_runner():
+            self.get_success_response(self.organization.slug, self.workflow.id)
+
+        # Ensure the workflow is scheduled for deletion
+        assert RegionScheduledDeletion.objects.filter(
+            model_name="Workflow",
+            object_id=self.workflow.id,
+        ).exists()
+
+        # Actually delete the workflow
+        with self.tasks():
+            run_scheduled_deletions()
+
+        assert not DataConditionGroup.objects.filter(id=action_condition_group.id).exists()
+
+    def test_without_permissions(self):
+        # Create a workflow with a different organization
+        new_org = self.create_organization()
+        workflow = self.create_workflow(organization_id=new_org.id)
+
+        with outbox_runner():
+            response = self.get_error_response(self.organization.slug, workflow.id)
+            assert response.status_code == 404