studio/people
Archived
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

✨(admin) send pending mailboxes from admin

Browse Source 
Provides an admin action to send all pending mailboxes for an active domain.
This allows quick fixes when mailboxes fell out of sync.
This commit is contained in:
Marie PUPO JEAMMET
2025-05-07 11:42:13 +02:00
committed by Marie
parent fe9fb67fed
commit 6e792986be
4 changed files with 141 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
src/backend/mailbox_manager/admin.py
View File

@@ -135,6 +135,39 @@ def fetch_domain_expected_config_from_dimail(modeladmin, request, queryset): #
)
@admin.action(description=_("Send pending mailboxes to dimail"))
def send_pending_mailboxes(modeladmin, request, queryset): # pylint: disable=unused-argument
"""Send pending mailboxes"""
client = DimailAPIClient()
excluded_domains = []
for domain in queryset:
# do not check disabled domains
if domain.status != enums.MailDomainStatusChoices.ENABLED:
excluded_domains.append(domain.name)
continue
results = client.send_pending_mailboxes(domain)
if failed_mailboxes := results["failed_mailboxes"]:
messages.error(
request,
_("Failed to send the following mailboxes : %(mailboxes)s.")
% {"mailboxes": ", ".join(failed_mailboxes)},
)
else:
messages.success(
request,
_("Pending mailboxes successfully sent for %(domain)s.")
% {"domain": domain.name},
)
if excluded_domains:
messages.warning(
request,
_("Domains disabled are excluded from : %(domains)s")
% {"domains": ", ".join(excluded_domains)},
)
class UserMailDomainAccessInline(admin.TabularInline):
"""Inline admin class for mail domain accesses."""
@@ -163,6 +196,7 @@ class MailDomainAdmin(admin.ModelAdmin):
sync_mailboxes_from_dimail,
fetch_domain_status_from_dimail,
fetch_domain_expected_config_from_dimail,
send_pending_mailboxes,
)
autocomplete_fields = ["organization"]

80
src/backend/mailbox_manager/tests/test_admin_actions.py
View File

@@ -204,3 +204,83 @@ def test_fetch_domain_expected_config__should_not_fetch_for_disabled_domain(clie
assert "Domains disabled are excluded from fetch" in response.content.decode(
"utf-8"
)
@responses.activate
@pytest.mark.django_db
def test_send_pending_mailboxes(client):
"""Test admin action to send pending mailboxes to dimail."""
admin = core_factories.UserFactory(is_staff=True, is_superuser=True)
client.force_login(admin)
domain = factories.MailDomainFactory(status=enums.MailDomainStatusChoices.ENABLED)
mailboxes = factories.MailboxFactory.create_batch(
3, status=enums.MailboxStatusChoices.PENDING, domain=domain
)
data = {
"action": "send_pending_mailboxes",
"_selected_action": [domain.id],
}
url = reverse("admin:mailbox_manager_maildomain_changelist")
for mailbox in mailboxes:
responses.add(
responses.GET,
re.compile(r".*/token/"),
body=TOKEN_OK,
status=status.HTTP_200_OK,
content_type="application/json",
)
responses.add(
responses.POST,
re.compile(rf".*/domains/{domain.name}/mailboxes/"),
body=response_mailbox_created(f"{mailbox.local_part}@{domain.name}"),
status=status.HTTP_201_CREATED,
content_type="application/json",
)
response = client.post(url, data, follow=True)
assert response.status_code == status.HTTP_200_OK
for mailbox in mailboxes:
mailbox.refresh_from_db()
assert mailbox.status == enums.MailboxStatusChoices.ENABLED
@responses.activate
@pytest.mark.django_db
def test_send_pending_mailboxes__listing_failed_mailboxes(client):
"""Test admin action to send pending mailboxes to dimail."""
admin = core_factories.UserFactory(is_staff=True, is_superuser=True)
client.force_login(admin)
domain = factories.MailDomainFactory(status=enums.MailDomainStatusChoices.ENABLED)
mailbox = factories.MailboxFactory(
status=enums.MailboxStatusChoices.PENDING, domain=domain
)
data = {
"action": "send_pending_mailboxes",
"_selected_action": [domain.id],
}
url = reverse("admin:mailbox_manager_maildomain_changelist")
responses.add(
responses.GET,
re.compile(r".*/token/"),
body=TOKEN_OK,
status=status.HTTP_200_OK,
content_type="application/json",
)
responses.add(
responses.POST,
re.compile(rf".*/domains/{domain.name}/mailboxes/"),
body=response_mailbox_created(f"{mailbox.local_part}@{domain.name}"),
status=status.HTTP_409_CONFLICT,
content_type="application/json",
)
response = client.post(url, data, follow=True)
assert response.status_code == status.HTTP_200_OK
assert (
f"Failed to send the following mailboxes : {str(mailbox)}"
in response.content.decode("utf-8")
)
mailbox.refresh_from_db()
assert mailbox.status == enums.MailboxStatusChoices.PENDING

6
src/backend/mailbox_manager/tests/test_utils_dimail_client.py
View File

@@ -348,9 +348,9 @@ def test_dimail__fetch_domain_status__full_fix_scenario(domain_status):
assert domain.last_check_details == body_domain_ok
def test_dimail__enable_pending_mailboxes(caplog):
def test_dimail__send_pending_mailboxes(caplog):
"""Status of pending mailboxes should switch to "enabled"
when calling enable_pending_mailboxes."""
when calling send_pending_mailboxes."""
caplog.set_level(logging.INFO)
domain = factories.MailDomainFactory()
@@ -380,7 +380,7 @@ def test_dimail__enable_pending_mailboxes(caplog):
status=status.HTTP_201_CREATED,
content_type="application/json",
)
dimail_client.enable_pending_mailboxes(domain=domain)
dimail_client.send_pending_mailboxes(domain=domain)
mailbox1.refresh_from_db()
mailbox2.refresh_from_db()

43
src/backend/mailbox_manager/utils/dimail.py
View File

@@ -427,29 +427,34 @@ class DimailAPIClient:
return response
return self.raise_exception_for_unexpected_response(response)
def enable_pending_mailboxes(self, domain):
"""Send requests for all pending mailboxes of a domain."""
def send_pending_mailboxes(self, domain):
"""Send requests for all pending mailboxes of a domain. Returns a list of failed mailboxes for this domain."""
failed_mailboxes = []
for mailbox in domain.mailboxes.filter(
status=enums.MailboxStatusChoices.PENDING
):
response = self.create_mailbox(mailbox)
mailbox.status = enums.MailDomainStatusChoices.ENABLED
mailbox.save()
if mailbox.secondary_email:
# send confirmation email
self.notify_mailbox_creation(
recipient=mailbox.secondary_email,
mailbox_data=response.json(),
)
try:
response = self.create_mailbox(mailbox)
except requests.exceptions.HTTPError:
failed_mailboxes.append(str(mailbox))
else:
logger.warning(
"Email notification for %s creation not sent "
"because no secondary email found",
mailbox,
)
mailbox.status = enums.MailDomainStatusChoices.ENABLED
mailbox.save()
if mailbox.secondary_email and mailbox.secondary_email != str(mailbox):
# send confirmation email
self.notify_mailbox_creation(
recipient=mailbox.secondary_email,
mailbox_data=response.json(),
)
else:
logger.warning(
"Email notification for %s creation not sent "
"because no valid secondary email found",
mailbox,
)
return {"failed_mailboxes": failed_mailboxes}
def check_domain(self, domain):
"""Send a request to dimail to check domain health."""
@@ -498,7 +503,7 @@ class DimailAPIClient:
domain.status != enums.MailDomainStatusChoices.ENABLED
and dimail_state == "ok"
):
self.enable_pending_mailboxes(domain)
self.send_pending_mailboxes(domain)
domain.status = enums.MailDomainStatusChoices.ENABLED
domain.last_check_details = dimail_response
domain.save()