Files
prowler/api/src/backend/tasks/tests/test_deletion.py
2026-02-05 10:29:37 +01:00

145 lines
5.4 KiB
Python

from unittest.mock import call, patch
import pytest
from django.core.exceptions import ObjectDoesNotExist
from api.models import Provider, Tenant
from tasks.jobs.deletion import delete_provider, delete_tenant
@pytest.mark.django_db
class TestDeleteProvider:
def test_delete_provider_success(self, providers_fixture):
with (
patch(
"tasks.jobs.deletion.graph_database.get_database_name",
return_value="tenant-db",
) as mock_get_database_name,
patch(
"tasks.jobs.deletion.graph_database.drop_subgraph"
) as mock_drop_subgraph,
):
instance = providers_fixture[0]
tenant_id = str(instance.tenant_id)
result = delete_provider(tenant_id, instance.id)
assert result
with pytest.raises(ObjectDoesNotExist):
Provider.objects.get(pk=instance.id)
mock_get_database_name.assert_called_once_with(tenant_id)
mock_drop_subgraph.assert_called_once_with(
"tenant-db",
str(instance.id),
)
def test_delete_provider_does_not_exist(self, tenants_fixture):
with (
patch(
"tasks.jobs.deletion.graph_database.get_database_name",
return_value="tenant-db",
) as mock_get_database_name,
patch(
"tasks.jobs.deletion.graph_database.drop_subgraph"
) as mock_drop_subgraph,
):
tenant_id = str(tenants_fixture[0].id)
non_existent_pk = "babf6796-cfcc-4fd3-9dcf-88d012247645"
with pytest.raises(ObjectDoesNotExist):
delete_provider(tenant_id, non_existent_pk)
mock_get_database_name.assert_called_once_with(tenant_id)
mock_drop_subgraph.assert_called_once_with(
"tenant-db",
non_existent_pk,
)
@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.
"""
with (
patch(
"tasks.jobs.deletion.graph_database.get_database_name",
return_value="tenant-db",
) as mock_get_database_name,
patch(
"tasks.jobs.deletion.graph_database.drop_subgraph"
) as mock_drop_subgraph,
patch(
"tasks.jobs.deletion.graph_database.drop_database"
) as mock_drop_database,
):
tenant = tenants_fixture[0]
providers = list(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
# 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()
# get_database_name is called once per provider + once for drop_database
expected_get_db_calls = [call(tenant.id) for _ in providers] + [
call(tenant.id)
]
mock_get_database_name.assert_has_calls(
expected_get_db_calls, any_order=True
)
assert mock_get_database_name.call_count == len(expected_get_db_calls)
expected_drop_subgraph_calls = [
call("tenant-db", str(provider.id)) for provider in providers
]
mock_drop_subgraph.assert_has_calls(
expected_drop_subgraph_calls,
any_order=True,
)
assert mock_drop_subgraph.call_count == len(expected_drop_subgraph_calls)
mock_drop_database.assert_called_once_with("tenant-db")
def test_delete_tenant_with_no_providers(self, tenants_fixture):
"""
Test deletion of a tenant with no related providers.
"""
with (
patch(
"tasks.jobs.deletion.graph_database.get_database_name",
return_value="tenant-db",
) as mock_get_database_name,
patch(
"tasks.jobs.deletion.graph_database.drop_subgraph"
) as mock_drop_subgraph,
patch(
"tasks.jobs.deletion.graph_database.drop_database"
) as mock_drop_database,
):
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()
# get_database_name is called once for drop_database
mock_get_database_name.assert_called_once_with(tenant.id)
mock_drop_subgraph.assert_not_called()
mock_drop_database.assert_called_once_with("tenant-db")