(aliases) delete all aliases of a given local part

added a bulk delete method for aliases, when filtering on local part
this is convenient when in need to delete the local part and all its
destinations in a single call
This commit is contained in:
Marie PUPO JEAMMET
2026-01-09 15:55:08 +01:00
committed by Marie
parent 8ab1b2e2ef
commit bc1cbef168
4 changed files with 178 additions and 1 deletions

View File

@@ -8,6 +8,7 @@ and this project adheres to
## [Unreleased]
- ✨(aliases) delete all aliases in one call #1002
- ✨(aliases) fix deleting single aliases #1002
- 🔥(plugins) remove CommuneCreation plugin

View File

@@ -425,6 +425,9 @@ class AliasViewSet(
DELETE /api/<version>/mail-domains/<domain_slug>/aliases/<alias_pk>/
Delete targeted alias
DELETE /api/<version>/mail-domains/<domain_slug>/aliases/?local_part=<local_part>/
Delete all aliases of targeted local_part
"""
lookup_field = "pk"
@@ -477,3 +480,26 @@ class AliasViewSet(
)
return Response(status=status.HTTP_204_NO_CONTENT)
@action(methods=["DELETE"], detail=False)
def delete(self, request, *args, **kwargs):
"""Bulk delete aliases. Filtering is required and accepted filter is local_part."""
if "local_part" not in self.request.query_params:
return Response(status=status.HTTP_400_BAD_REQUEST)
local_part = self.request.query_params["local_part"]
queryset = self.get_queryset().filter(
local_part=local_part
) # Manually call get_queryset to filter by domain and role
if not queryset:
raise Http404("No Alias matches the given query.")
# view is bounded to a domain, fetch is from the queryset to spare a dedicated DB request"
domain_name = queryset[0].domain.name
queryset.delete()
client = DimailAPIClient()
client.delete_multiple_alias(local_part, domain_name)
return Response(status=status.HTTP_204_NO_CONTENT)

View File

@@ -0,0 +1,129 @@
"""
Tests for aliases API endpoint in People's app mailbox_manager.
Focus on "bulk delete" action.
"""
# pylint: disable=W0613
import re
import pytest
import responses
from rest_framework import status
from rest_framework.test import APIClient
from core import factories as core_factories
from mailbox_manager import enums, factories, models
pytestmark = pytest.mark.django_db
def test_api_aliases_bulk_delete__anonymous_get_401():
"""Anonymous user should not be able to bulk delete."""
mail_domain = factories.MailDomainFactory()
alias_, _, _ = factories.AliasFactory.create_batch(3, domain=mail_domain)
client = APIClient()
response = client.delete(
f"/api/v1.0/mail-domains/{mail_domain.slug}/aliases/?local_part={alias_.local_part}",
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert models.Alias.objects.count() == 3
def test_api_aliases_bulk_delete__no_access_get_404():
"""User with no access to domain should not be able to bulk delete."""
mail_domain = factories.MailDomainFactory()
alias_, _, _ = factories.AliasFactory.create_batch(3, domain=mail_domain)
client = APIClient()
client.force_login(core_factories.UserFactory())
response = client.delete(
f"/api/v1.0/mail-domains/{mail_domain.slug}/aliases/?local_part={alias_.local_part}",
)
assert response.status_code == status.HTTP_404_NOT_FOUND
assert models.Alias.objects.count() == 3
def test_api_aliases_bulk_delete__viewer_get_403():
"""Viewer user should not be able to bulk delete."""
access = factories.MailDomainAccessFactory(role=enums.MailDomainRoleChoices.VIEWER)
alias_, _, _ = factories.AliasFactory.create_batch(3, domain=access.domain)
client = APIClient()
client.force_login(access.user)
response = client.delete(
f"/api/v1.0/mail-domains/{access.domain.slug}/aliases/?local_part={alias_.local_part}",
)
assert response.status_code == status.HTTP_403_FORBIDDEN
assert models.Alias.objects.count() == 3
@responses.activate
def test_api_aliases_bulk_delete__administrators_allowed_all_destination(
dimail_token_ok,
):
"""
Administrators of a domain should be allowed to bulk delete all aliases
of a given local_part.
"""
authenticated_user = core_factories.UserFactory()
mail_domain = factories.MailDomainFactory(
users=[(authenticated_user, enums.MailDomainRoleChoices.ADMIN)]
)
alias_ = factories.AliasFactory(domain=mail_domain)
factories.AliasFactory.create_batch(
2, domain=mail_domain, local_part=alias_.local_part
)
# additional aliases that shouldn't be affected
factories.AliasFactory.create_batch(
2, domain=mail_domain, destination=alias_.destination
)
factories.AliasFactory(
local_part=alias_.local_part,
destination=alias_.destination,
)
# Mock dimail response
responses.delete(
re.compile(r".*/aliases/"),
status=status.HTTP_204_NO_CONTENT,
content_type="application/json",
)
client = APIClient()
client.force_login(authenticated_user)
response = client.delete(
f"/api/v1.0/mail-domains/{mail_domain.slug}/aliases/?local_part={alias_.local_part}",
)
assert response.status_code == status.HTTP_204_NO_CONTENT
assert models.Alias.objects.count() == 3
assert not models.Alias.objects.filter(
domain=mail_domain, local_part=alias_.local_part
).exists()
def test_api_aliases_bulk_delete__no_local_part_bad_request():
"""Filtering by local part is mandatory when bulk deleting aliases."""
authenticated_user = core_factories.UserFactory()
mail_domain = factories.MailDomainFactory(
users=[(authenticated_user, enums.MailDomainRoleChoices.ADMIN)]
)
alias_ = factories.AliasFactory(domain=mail_domain)
factories.AliasFactory.create_batch(
2, domain=mail_domain, local_part=alias_.local_part
)
client = APIClient()
client.force_login(authenticated_user)
response = client.delete(
f"/api/v1.0/mail-domains/{mail_domain.slug}/aliases/",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert models.Alias.objects.count() == 3

View File

@@ -778,11 +778,32 @@ class DimailAPIClient:
str(alias.domain),
)
# we don't raise error because we actually want this alias to be deleted
# to match dimail's states
return response
return self._raise_exception_for_unexpected_response(response)
def delete_multiple_alias(self, local_part, domain_name):
"""Send a Delete alias request to mail provisioning API."""
try:
response = session.delete(
f"{self.API_URL}/domains/{domain_name}/aliases/{local_part}/all",
json={},
headers=self._get_headers(),
verify=True,
timeout=self.API_TIMEOUT,
)
except requests.exceptions.ConnectionError as error:
logger.error(
"Connection error while trying to reach %s.",
self.API_URL,
exc_info=error,
)
raise error
# response.raise_for_status()
return response
def import_aliases(self, domain):
"""Import aliases from dimail. Useful if people fall out of sync with dimail."""