✨(dimail) management command to fetch domain status

CHANGELOG.md

@@ -8,6 +8,10 @@ and this project adheres to
 
 ## [Unreleased]
 
+### Added
+
+- ✨(dimail) management command to fetch domain status
+
 ### Changed
 
 - ✨(scripts) adapts release script after moving the deployment part

Makefile

@@ -403,3 +403,7 @@ install-secret: ## install the kubernetes secrets from Vaultwarden
        --set installCRDs=true; \
 	fi
 .PHONY: build-k8s-cluster
+
+fetch-domain-status:
+	@$(MANAGE) fetch_domain_status
+.PHONY: fetch-domain-status

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

@@ -0,0 +1,58 @@
+"""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("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"
+                domain_name = (
+                    f"{domain.name[:40]}..." if len(domain.name) > 40 else domain.name
+                )
+                self.stdout.write(
+                    self.style.SUCCESS(
+                        (
+                            f"Domain {domain_name}"
+                            + "." * (50 - len(domain_name))
+                            + action
+                        )
+                    )
+                )
+        self.stdout.write("Done", ending="\n")

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

@@ -0,0 +1,69 @@
+"""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
+
+
+@responses.activate
+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
+    )
+
+    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),
+    ]:
+        # mock dimail API
+        responses.add(
+            responses.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()
+    # nothing change for the first domain enable
+    assert domain_enabled1.status == enums.MailDomainStatusChoices.ENABLED
+    # status of the second activated domain has changed to failure
+    assert domain_enabled2.status == enums.MailDomainStatusChoices.FAILED
+    # status of the failed domain has changed to enabled
+    assert domain_failed.status == enums.MailDomainStatusChoices.ENABLED
+    # disabled domain was excluded
+    assert domain_disabled.status == enums.MailDomainStatusChoices.DISABLED
+    assert output.getvalue().count("CHECKED") == 1
+    assert output.getvalue().count("UPDATED") == 2