fix(tenant): fix delete tenants behavior (#6013)

api/src/backend/tasks/jobs/deletion.py

@@ -1,8 +1,9 @@
 from celery.utils.log import get_task_logger
 from django.db import transaction
 
-from api.db_utils import batch_delete
-from api.models import Finding, Provider, Resource, Scan, ScanSummary
+from api.db_router import MainRouter
+from api.db_utils import batch_delete, tenant_transaction
+from api.models import Finding, Provider, Resource, Scan, ScanSummary, Tenant
 
 logger = get_task_logger(__name__)
 
@@ -49,3 +50,26 @@ def delete_provider(pk: str):
         deletion_summary.update(provider_summary)
 
     return deletion_summary
+
+
+def delete_tenant(pk: str):
+    """
+    Gracefully deletes an instance of a tenant along with its related data.
+
+    Args:
+        pk (str): The primary key of the Tenant instance to delete.
+
+    Returns:
+        dict: A dictionary with the count of deleted objects per model,
+              including related models.
+    """
+    deletion_summary = {}
+
+    for provider in Provider.objects.using(MainRouter.admin_db).filter(tenant_id=pk):
+        with tenant_transaction(pk):
+            summary = delete_provider(provider.id)
+            deletion_summary.update(summary)
+
+    Tenant.objects.using(MainRouter.admin_db).filter(id=pk).delete()
+
+    return deletion_summary

api/src/backend/tasks/tasks.py

@@ -4,7 +4,7 @@ from celery import shared_task
 from config.celery import RLSTask
 from django_celery_beat.models import PeriodicTask
 from tasks.jobs.connection import check_provider_connection
-from tasks.jobs.deletion import delete_provider
+from tasks.jobs.deletion import delete_provider, delete_tenant
 from tasks.jobs.scan import aggregate_findings, perform_prowler_scan
 
 from api.db_utils import tenant_transaction
@@ -134,3 +134,8 @@ def perform_scheduled_scan_task(self, tenant_id: str, provider_id: str):
 @shared_task(name="scan-summary")
 def perform_scan_summary_task(tenant_id: str, scan_id: str):
     return aggregate_findings(tenant_id=tenant_id, scan_id=scan_id)
+
+
+@shared_task(name="tenant-deletion")
+def delete_tenant_task(tenant_id: str):
+    return delete_tenant(pk=tenant_id)

api/src/backend/tasks/tests/test_deletion.py

@@ -1,13 +1,15 @@
+from unittest.mock import patch
+
 import pytest
 from django.core.exceptions import ObjectDoesNotExist
-from tasks.jobs.deletion import delete_provider
+from tasks.jobs.deletion import delete_provider, delete_tenant
 
-from api.models import Provider
+from api.models import Provider, Tenant
 
 
 @pytest.mark.django_db
-class TestDeleteInstance:
-    def test_delete_instance_success(self, providers_fixture):
+class TestDeleteProvider:
+    def test_delete_provider_success(self, providers_fixture):
         instance = providers_fixture[0]
         result = delete_provider(instance.id)
 
@@ -15,8 +17,47 @@ class TestDeleteInstance:
         with pytest.raises(ObjectDoesNotExist):
             Provider.objects.get(pk=instance.id)
 
-    def test_delete_instance_does_not_exist(self):
+    def test_delete_provider_does_not_exist(self):
         non_existent_pk = "babf6796-cfcc-4fd3-9dcf-88d012247645"
 
         with pytest.raises(ObjectDoesNotExist):
             delete_provider(non_existent_pk)
+
+
+@patch("api.db_router.MainRouter.admin_db", new="default")
+@pytest.mark.django_db
+class TestDeleteTenant:
+    def test_delete_tenant_success(self, tenants_fixture, providers_fixture):
+        """
+        Test successful deletion of a tenant and its related data.
+        """
+        tenant = tenants_fixture[0]
+        providers = Provider.objects.filter(tenant_id=tenant.id)
+
+        # Ensure the tenant and related providers exist before deletion
+        assert Tenant.objects.filter(id=tenant.id).exists()
+        assert providers.exists()
+
+        # Call the function and validate the result
+        deletion_summary = delete_tenant(tenant.id)
+
+        assert deletion_summary is not None
+        assert not Tenant.objects.filter(id=tenant.id).exists()
+        assert not Provider.objects.filter(tenant_id=tenant.id).exists()
+
+    def test_delete_tenant_with_no_providers(self, tenants_fixture):
+        """
+        Test deletion of a tenant with no related providers.
+        """
+        tenant = tenants_fixture[1]  # Assume this tenant has no providers
+        providers = Provider.objects.filter(tenant_id=tenant.id)
+
+        # Ensure the tenant exists but has no related providers
+        assert Tenant.objects.filter(id=tenant.id).exists()
+        assert not providers.exists()
+
+        # Call the function and validate the result
+        deletion_summary = delete_tenant(tenant.id)
+
+        assert deletion_summary == {}  # No providers, so empty summary
+        assert not Tenant.objects.filter(id=tenant.id).exists()