✨(backend) add admin action to check domain health

Allow to select some domains to check and update status
thanks to a dimail call.

src/backend/mailbox_manager/admin.py

@@ -1,11 +1,12 @@
 """Admin classes and registrations for People's mailbox manager app."""
 
 from django.contrib import admin, messages
+from django.utils.html import format_html
 from django.utils.translation import gettext_lazy as _
 
 from requests import exceptions
 
-from mailbox_manager import models
+from mailbox_manager import enums, models
 from mailbox_manager.utils.dimail import DimailAPIClient
 
 
@@ -32,6 +33,51 @@ def sync_mailboxes_from_dimail(modeladmin, request, queryset):  # pylint: disabl
             )
 
 
+@admin.action(description=_("Check and update status from dimail"))
+def fetch_domain_status_from_dimail(modeladmin, request, queryset):  # pylint: disable=unused-argument
+    """Admin action to check domain health with dimail and update domain status."""
+    client = DimailAPIClient()
+    domains_updated, excluded_domains, msg_error = [], [], []
+    success = False
+    for domain in queryset:
+        # do not check disabled domains
+        if domain.status == enums.MailDomainStatusChoices.DISABLED:
+            excluded_domains.append(domain.name)
+            continue
+
+        old_status = domain.status
+        try:
+            response = client.fetch_domain_status(domain)
+        except exceptions.HTTPError as err:
+            msg_error.append(_(f"""- <b>{domain.name}</b> with message: '{err}'"""))
+        else:
+            success = True
+            # temporary (or not?) display content of the dimail response to debug broken state
+            if domain.status == enums.MailDomainStatusChoices.FAILED:
+                messages.info(request, response.json())
+            if old_status != domain.status:
+                domains_updated.append(domain.name)
+
+    if success:
+        msg_success = [
+            _("Check domains done with success."),
+            _(f"Domains updated: {', '.join(domains_updated)}")
+            if domains_updated
+            else _("No domain updated."),
+        ]
+        messages.success(request, format_html("<br> ".join(map(str, msg_success))))
+    if msg_error:
+        msg_error.insert(0, _("Check domain failed for:"))
+        messages.error(request, format_html("<br> ".join(map(str, msg_error))))
+    if excluded_domains:
+        messages.warning(
+            request,
+            _(
+                f"Domains disabled are excluded from check: {', '.join(excluded_domains)}"
+            ),
+        )
+
+
 class UserMailDomainAccessInline(admin.TabularInline):
     """Inline admin class for mail domain accesses."""
 
@@ -54,7 +100,7 @@ class MailDomainAdmin(admin.ModelAdmin):
     search_fields = ("name",)
     readonly_fields = ["created_at", "slug"]
     inlines = (UserMailDomainAccessInline,)
-    actions = (sync_mailboxes_from_dimail,)
+    actions = (sync_mailboxes_from_dimail, fetch_domain_status_from_dimail)
 
 
 @admin.register(models.Mailbox)

src/backend/mailbox_manager/tests/test_admin_actions.py

@@ -0,0 +1,78 @@
+"""
+Unit tests for admin actions
+"""
+
+import json
+import re
+
+from django.urls import reverse
+
+import pytest
+import responses
+
+from core import factories as core_factories
+
+from mailbox_manager import enums, factories
+
+from .fixtures.dimail import CHECK_DOMAIN_BROKEN
+
+
+@pytest.mark.django_db
+def test_admin_action__fetch_domain_status_from_dimail(client):
+    """Test admin action to check health of some domains"""
+    admin = core_factories.UserFactory(is_staff=True, is_superuser=True)
+    client.force_login(admin)
+    domain1 = factories.MailDomainEnabledFactory()
+    domain2 = factories.MailDomainEnabledFactory()
+    data = {
+        "action": "fetch_domain_status_from_dimail",
+        "_selected_action": [
+            domain1.id,
+            domain2.id,
+        ],
+    }
+    url = reverse("admin:mailbox_manager_maildomain_changelist")
+
+    with responses.RequestsMock() as rsps:
+        body_content_domain1 = CHECK_DOMAIN_BROKEN.copy()
+        body_content_domain1["name"] = domain1.name
+        body_content_domain2 = CHECK_DOMAIN_BROKEN.copy()
+        body_content_domain2["name"] = domain2.name
+        rsps.add(
+            rsps.GET,
+            re.compile(rf".*/domains/{domain1.name}/check/"),
+            body=json.dumps(body_content_domain1),
+            status=200,
+            content_type="application/json",
+        )
+        rsps.add(
+            rsps.GET,
+            re.compile(rf".*/domains/{domain2.name}/check/"),
+            body=json.dumps(body_content_domain2),
+            status=200,
+            content_type="application/json",
+        )
+        response = client.post(url, data, follow=True)
+        assert response.status_code == 200
+        domain1.refresh_from_db()
+        domain2.refresh_from_db()
+        assert domain1.status == enums.MailDomainStatusChoices.FAILED
+        assert domain2.status == enums.MailDomainStatusChoices.FAILED
+        assert "Check domains done with success" in response.content.decode("utf-8")
+
+        # check with a valid domain info from dimail
+        body_content_domain1["state"] = "ok"
+        rsps.add(
+            rsps.GET,
+            re.compile(rf".*/domains/{domain1.name}/check/"),
+            body=json.dumps(body_content_domain1),
+            status=200,
+            content_type="application/json",
+        )
+        response = client.post(url, data, follow=True)
+        assert response.status_code == 200
+        domain1.refresh_from_db()
+        domain2.refresh_from_db()
+        assert domain1.status == enums.MailDomainStatusChoices.ENABLED
+        assert domain2.status == enums.MailDomainStatusChoices.FAILED
+        assert "Check domains done with success" in response.content.decode("utf-8")