✨(dimail) management command to fetch domain status

Add a management command for a future cron to
to check and update regularly domains status
from dimail.

CHANGELOG.md

@@ -10,6 +10,7 @@ and this project adheres to
 
 ### Added
 
+- ✨(dimail) management command to fetch domain status
 - ✨(organization) add admin action for plugin #640
 - ✨(anct) fetch and display organization names of communes #583
 - ✨(frontend) display email if no username #562

Makefile

@@ -361,3 +361,7 @@ tilt-up: ## start tilt - k8s local development
 release:  ## helper for release and deployment
 	python scripts/release.py
 .PHONY: release
+
+fetch-domain-status:
+	@$(MANAGE) fetch_domain_status
+.PHONY: fetch-domain-status

src/backend/mailbox_manager/management/commands/fetch_domain_status.py

@@ -0,0 +1,57 @@
+"""Management command to check and update domain status"""
+
+import logging
+
+from django.core.management.base import BaseCommand
+
+import requests
+
+from mailbox_manager.enums import MailDomainStatusChoices
+from mailbox_manager.models import MailDomain
+from mailbox_manager.utils.dimail import DimailAPIClient
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    """
+    Management command to check and update domains status from dimail
+    """
+
+    help = (
+        "This command calls dimail to get and update the status of domains."
+        "All domains without a disabled status will be checked and updated if status"
+        "sent by dimail does not match our status saved in our database."
+    )
+
+    def handle(self, *args, **options):
+        """Handling of the management command."""
+
+        self.stdout.write(
+            self.style.SUCCESS("Start fetching domain status from dimail...")
+        )
+        client = DimailAPIClient()
+        # do not fetch status of disabled domains
+        domains = MailDomain.objects.exclude(status=MailDomainStatusChoices.DISABLED)
+        for domain in domains:
+            old_status = domain.status
+            try:
+                client.fetch_domain_status(domain)
+            except requests.exceptions.HTTPError as err:
+                self.stdout.write(
+                    self.style.ERROR(
+                        f"Fetch failed for {domain.name} with message: '{err}'"
+                    )
+                )
+            else:
+                action = "updated" if old_status != domain.status else "checked"
+                self.stdout.write(
+                    self.style.SUCCESS(
+                        (
+                            f"Domain {domain.name} {action}"
+                            + "." * (50 - len(domain.name))
+                            + "OK"
+                        )
+                    )
+                )
+        self.stdout.write("DONE", ending="\n")

src/backend/mailbox_manager/tests/commands/test_fetch_domain_status.py

@@ -0,0 +1,64 @@
+"""Test the `fetch_domain_status_from_dimail` management command"""
+
+import json
+import re
+from io import StringIO
+
+from django.core.management import call_command
+
+import pytest
+import responses
+
+from mailbox_manager import enums, factories
+from mailbox_manager.tests.fixtures.dimail import CHECK_DOMAIN_BROKEN, CHECK_DOMAIN_OK
+
+pytestmark = pytest.mark.django_db
+
+
+def test_fetch_domain_status():
+    """Test fetch domain status from dimail"""
+    domain_enabled1 = factories.MailDomainEnabledFactory()
+    domain_enabled2 = factories.MailDomainEnabledFactory()
+    domain_disabled = factories.MailDomainFactory(
+        status=enums.MailDomainStatusChoices.DISABLED
+    )
+    domain_failed = factories.MailDomainFactory(
+        status=enums.MailDomainStatusChoices.FAILED
+    )
+
+    with responses.RequestsMock() as rsps:
+        body_content_ok1 = CHECK_DOMAIN_OK.copy()
+        body_content_ok1["name"] = domain_enabled1.name
+
+        body_content_broken = CHECK_DOMAIN_BROKEN.copy()
+        body_content_broken["name"] = domain_enabled2.name
+
+        body_content_ok2 = CHECK_DOMAIN_OK.copy()
+        body_content_ok2["name"] = domain_disabled.name
+
+        body_content_ok3 = CHECK_DOMAIN_OK.copy()
+        body_content_ok3["name"] = domain_failed.name
+        for domain, body_content in [
+            (domain_enabled1, body_content_ok1),
+            (domain_enabled2, body_content_broken),
+            (domain_failed, body_content_ok3),
+        ]:
+            rsps.add(
+                rsps.GET,
+                re.compile(rf".*/domains/{domain.name}/check/"),
+                body=json.dumps(body_content),
+                status=200,
+                content_type="application/json",
+            )
+        output = StringIO()
+        call_command("fetch_domain_status", verbosity=2, stdout=output)
+        domain_enabled1.refresh_from_db()
+        domain_enabled2.refresh_from_db()
+        domain_disabled.refresh_from_db()
+        domain_failed.refresh_from_db()
+        assert domain_enabled1.status == enums.MailDomainStatusChoices.ENABLED
+        assert domain_enabled2.status == enums.MailDomainStatusChoices.FAILED
+        # disabled domain was excluded
+        assert domain_disabled.status == enums.MailDomainStatusChoices.DISABLED
+        assert domain_failed.status == enums.MailDomainStatusChoices.ENABLED
+        assert output.getvalue().count("OK") == 3