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

✨(dimail) manage 'action required' status for MailDomain

Browse Source 
Adapt fetch domain status call to manage internal and external
fixes required. Use the new status 'action required' to
manage actions expected from support.
Call a new dimail endpoint to run a fix for internal checks
when all external checks are OK.
This commit is contained in:
Sabrina Demagny
2025-02-04 13:26:15 +01:00
parent e7af1fd591
commit 4df4172151
6 changed files with 463 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -10,6 +10,7 @@ and this project adheres to
### Added
- ✨(dimail) manage 'action required' status for MailDomain
- ✨(domains) add action required status on MailDomain
- ✨(dimail) send pending mailboxes upon domain activation

12
src/backend/mailbox_manager/tests/commands/test_fetch_domain_status.py
View File

@@ -8,9 +8,13 @@ from django.core.management import call_command
import pytest
import responses
from rest_framework import status
from mailbox_manager import enums, factories
from mailbox_manager.tests.fixtures.dimail import CHECK_DOMAIN_BROKEN, CHECK_DOMAIN_OK
from mailbox_manager.tests.fixtures.dimail import (
CHECK_DOMAIN_BROKEN,
CHECK_DOMAIN_OK,
)
pytestmark = pytest.mark.django_db
@@ -48,7 +52,7 @@ def test_fetch_domain_status():
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=200,
status=status.HTTP_200_OK,
content_type="application/json",
)
output = StringIO()
@@ -59,8 +63,8 @@ def test_fetch_domain_status():
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 second activated domain has changed to action required
assert domain_enabled2.status == enums.MailDomainStatusChoices.ACTION_REQUIRED
# status of the failed domain has changed to enabled
assert domain_failed.status == enums.MailDomainStatusChoices.ENABLED
# disabled domain was excluded

153
src/backend/mailbox_manager/tests/fixtures/dimail.py vendored
View File

@@ -16,7 +16,11 @@ CHECK_DOMAIN_BROKEN = {
"smtp_domain": None,
"context_name": "example.fr",
"transport": None,
"domain_exist": {"ok": True, "internal": False, "errors": []},
"domain_exist": {
"ok": True,
"internal": False,
"errors": [],
},
"mx": {
"ok": False,
"internal": False,
@@ -74,8 +78,16 @@ CHECK_DOMAIN_BROKEN = {
{"code": "no_dkim", "detail": "Il faut un DKIM record, avec la bonne clef"}
],
},
"postfix": {"ok": True, "internal": True, "errors": []},
"ox": {"ok": True, "internal": True, "errors": []},
"postfix": {
"ok": True,
"internal": True,
"errors": [],
},
"ox": {
"ok": True,
"internal": True,
"errors": [],
},
"cert": {
"ok": False,
"internal": True,
@@ -85,6 +97,141 @@ CHECK_DOMAIN_BROKEN = {
},
}
CHECK_DOMAIN_BROKEN_INTERNAL = {
"name": "example.fr",
"state": "broken",
"valid": False,
"delivery": "virtual",
"features": ["webmail", "mailbox"],
"webmail_domain": None,
"imap_domain": None,
"smtp_domain": None,
"context_name": "example.fr",
"transport": None,
"domain_exist": {
"ok": True,
"internal": False,
"errors": [],
},
"mx": {
"ok": True,
"internal": False,
"errors": [],
},
"cname_imap": {
"ok": True,
"internal": False,
"errors": [],
},
"cname_smtp": {
"ok": True,
"internal": False,
"errors": [],
},
"cname_webmail": {
"ok": True,
"internal": False,
"errors": [],
},
"spf": {
"ok": True,
"internal": False,
"errors": [],
},
"dkim": {
"ok": True,
"internal": False,
"errors": [],
},
"postfix": {
"ok": True,
"internal": True,
"errors": [],
},
"ox": {
"ok": True,
"internal": True,
"errors": [],
},
"cert": {
"ok": False,
"internal": True,
"errors": [
{"code": "no_cert", "detail": "Pas de certificat pour ce domaine (ls)"}
],
},
}
CHECK_DOMAIN_BROKEN_EXTERNAL = {
"name": "example.fr",
"state": "broken",
"valid": False,
"delivery": "virtual",
"features": ["webmail", "mailbox"],
"webmail_domain": None,
"imap_domain": None,
"smtp_domain": None,
"context_name": "example.fr",
"transport": None,
"domain_exist": {
"ok": True,
"internal": False,
"errors": [],
},
"mx": {
"ok": False,
"internal": False,
"errors": [
{
"code": "wrong_mx",
"detail": "Je veux que le MX du domaine soit mx.ox.numerique.gouv.fr., or je trouve example-fr.mail.protection.outlook.com.",
}
],
},
"cname_imap": {
"ok": True,
"internal": False,
"errors": [],
},
"cname_smtp": {
"ok": True,
"internal": False,
"errors": [],
},
"cname_webmail": {
"ok": True,
"internal": False,
"errors": [],
},
"spf": {
"ok": True,
"internal": False,
"errors": [],
},
"dkim": {
"ok": True,
"internal": False,
"errors": [],
},
"postfix": {
"ok": True,
"internal": True,
"errors": [],
},
"ox": {
"ok": True,
"internal": True,
"errors": [],
},
"cert": {
"ok": True,
"internal": True,
"errors": [],
},
}
CHECK_DOMAIN_OK = {
"name": "example.fr",
"state": "ok",

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

@@ -55,6 +55,7 @@ def test_sync_mailboxes__should_not_sync_if_domain_is_not_enabled(
)
@responses.activate
@pytest.mark.django_db
def test_fetch_domain_status__should_switch_to_failed_when_domain_broken(client):
"""Test admin action to check health of some domains"""
@@ -70,35 +71,34 @@ def test_fetch_domain_status__should_switch_to_failed_when_domain_broken(client)
],
}
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=status.HTTP_200_OK,
content_type="application/json",
)
rsps.add(
rsps.GET,
re.compile(rf".*/domains/{domain2.name}/check/"),
body=json.dumps(body_content_domain2),
status=status.HTTP_200_OK,
content_type="application/json",
)
response = client.post(url, data, follow=True)
assert response.status_code == status.HTTP_200_OK
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")
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
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain1.name}/check/"),
body=json.dumps(body_content_domain1),
status=status.HTTP_200_OK,
content_type="application/json",
)
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain2.name}/check/"),
body=json.dumps(body_content_domain2),
status=status.HTTP_200_OK,
content_type="application/json",
)
response = client.post(url, data, follow=True)
assert response.status_code == status.HTTP_200_OK
domain1.refresh_from_db()
domain2.refresh_from_db()
assert domain1.status == enums.MailDomainStatusChoices.ACTION_REQUIRED
assert domain2.status == enums.MailDomainStatusChoices.ACTION_REQUIRED
assert "Check domains done with success" in response.content.decode("utf-8")
@responses.activate
@pytest.mark.django_db
def test_fetch_domain_status__should_switch_to_enabled_when_domain_ok(client):
"""Test admin action should switch domain state to ENABLED
@@ -115,33 +115,33 @@ def test_fetch_domain_status__should_switch_to_enabled_when_domain_ok(client):
}
url = reverse("admin:mailbox_manager_maildomain_changelist")
with responses.RequestsMock() as rsps:
body_content_domain1 = CHECK_DOMAIN_OK.copy()
body_content_domain1["name"] = domain1.name
body_content_domain1 = CHECK_DOMAIN_OK.copy()
body_content_domain1["name"] = domain1.name
rsps.add(
rsps.GET,
re.compile(rf".*/domains/{domain1.name}/check/"),
body=json.dumps(body_content_domain1),
status=status.HTTP_200_OK,
content_type="application/json",
)
rsps.add(
rsps.GET,
re.compile(r".*/token/"),
body=TOKEN_OK,
status=status.HTTP_200_OK,
content_type="application/json",
)
rsps.add(
rsps.POST,
re.compile(rf".*/domains/{domain1.name}/mailboxes/"),
body=response_mailbox_created(f"truc@{domain1.name}"),
status=status.HTTP_201_CREATED,
content_type="application/json",
)
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain1.name}/check/"),
body=json.dumps(body_content_domain1),
status=status.HTTP_200_OK,
content_type="application/json",
)
# we need to get a token to create 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/{domain1.name}/mailboxes/"),
body=response_mailbox_created(f"truc@{domain1.name}"),
status=status.HTTP_201_CREATED,
content_type="application/json",
)
response = client.post(url, data, follow=True)
response = client.post(url, data, follow=True)
assert response.status_code == status.HTTP_200_OK
domain1.refresh_from_db()
domain2.refresh_from_db()

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

@@ -18,6 +18,8 @@ from mailbox_manager.utils.dimail import DimailAPIClient
from .fixtures.dimail import (
CHECK_DOMAIN_BROKEN,
CHECK_DOMAIN_BROKEN_EXTERNAL,
CHECK_DOMAIN_BROKEN_INTERNAL,
CHECK_DOMAIN_OK,
TOKEN_OK,
response_mailbox_created,
@@ -166,40 +168,180 @@ def test_dimail_synchronization__synchronize_mailboxes(mock_warning):
assert imported_mailboxes == [mailbox_valid["email"]]
def test_dimail__fetch_domain_status_from_dimail():
"""Request to dimail health status of a domain"""
domain = factories.MailDomainEnabledFactory()
with responses.RequestsMock() as rsps:
body_content = CHECK_DOMAIN_BROKEN.copy()
body_content["name"] = domain.name
rsps.add(
rsps.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client = DimailAPIClient()
response = dimail_client.fetch_domain_status(domain)
assert response.status_code == status.HTTP_200_OK
assert domain.status == enums.MailDomainStatusChoices.FAILED
@pytest.mark.parametrize(
"domain_status",
[
enums.MailDomainStatusChoices.PENDING,
enums.MailDomainStatusChoices.ACTION_REQUIRED,
enums.MailDomainStatusChoices.FAILED,
enums.MailDomainStatusChoices.ENABLED,
],
)
@responses.activate
def test_dimail__fetch_domain_status__switch_to_enabled(domain_status):
"""Domains should be enabled when dimail check returns ok status"""
domain = factories.MailDomainFactory(status=domain_status)
body_content = CHECK_DOMAIN_OK.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client = DimailAPIClient()
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.ENABLED
# Now domain is ok again
body_content = CHECK_DOMAIN_OK.copy()
body_content["name"] = domain.name
rsps.add(
rsps.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
response = dimail_client.fetch_domain_status(domain)
assert response.status_code == status.HTTP_200_OK
assert domain.status == enums.MailDomainStatusChoices.ENABLED
# call again, should be ok
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.ENABLED
def test_dimail___enable_pending_mailboxes(caplog):
@pytest.mark.parametrize(
"domain_status",
[
enums.MailDomainStatusChoices.PENDING,
enums.MailDomainStatusChoices.ENABLED,
enums.MailDomainStatusChoices.ACTION_REQUIRED,
enums.MailDomainStatusChoices.FAILED,
],
)
@responses.activate
def test_dimail__fetch_domain_status__switch_to_action_required(
domain_status,
):
"""Domains should be in status action required when dimail check
returns broken status for external checks."""
domain = factories.MailDomainFactory(status=domain_status)
body_content = CHECK_DOMAIN_BROKEN_EXTERNAL.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client = DimailAPIClient()
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.ACTION_REQUIRED
# Support team fixes their part of the problem
# Now domain is OK again
body_content = CHECK_DOMAIN_OK.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.ENABLED
@pytest.mark.parametrize(
"domain_status",
[
enums.MailDomainStatusChoices.PENDING,
enums.MailDomainStatusChoices.ENABLED,
enums.MailDomainStatusChoices.ACTION_REQUIRED,
],
)
@responses.activate
def test_dimail__fetch_domain_status__switch_to_failed(domain_status):
"""Domains should be in status failed when dimail check returns broken status
for only internal checks dispite a fix call."""
domain = factories.MailDomainFactory(status=domain_status)
# nothing can be done by support team, domain should be in failed
body_content = CHECK_DOMAIN_BROKEN_INTERNAL.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
# the endpoint fix is called and still returns KO for internal checks
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/fix/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client = DimailAPIClient()
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.FAILED
@pytest.mark.parametrize(
"domain_status",
[
enums.MailDomainStatusChoices.PENDING,
enums.MailDomainStatusChoices.ENABLED,
enums.MailDomainStatusChoices.ACTION_REQUIRED,
],
)
@responses.activate
def test_dimail__fetch_domain_status__full_fix_scenario(domain_status):
"""Domains should be enabled when dimail check returns ok status
after a fix call."""
domain = factories.MailDomainFactory(status=domain_status)
# with all checks KO, domain should be in action required
body_content = CHECK_DOMAIN_BROKEN.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client = DimailAPIClient()
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.ACTION_REQUIRED
# We assume that the support has fixed their part.
# So now dimail returns OK for external checks but still KO for internal checks.
# A call to dimail fix endpoint is necessary and will be done by
# the fetch_domain_status call
body_content = CHECK_DOMAIN_BROKEN_INTERNAL.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/check/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
# the endpoint fix is called and returns OK. Hooray!
body_content = CHECK_DOMAIN_OK.copy()
body_content["name"] = domain.name
responses.add(
responses.GET,
re.compile(rf".*/domains/{domain.name}/fix/"),
body=json.dumps(body_content),
status=status.HTTP_200_OK,
content_type="application/json",
)
dimail_client.fetch_domain_status(domain)
domain.refresh_from_db()
assert domain.status == enums.MailDomainStatusChoices.ENABLED
def test_dimail__enable_pending_mailboxes(caplog):
"""Status of pending mailboxes should switch to "enabled"
when calling enable_pending_mailboxes."""
caplog.set_level(logging.INFO)

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

@@ -399,32 +399,6 @@ class DimailAPIClient:
return response
return self.raise_exception_for_unexpected_response(response)
def fetch_domain_status(self, domain):
"""Send a request to check domain and update status of our domain."""
response = session.get(
f"{self.API_URL}/domains/{domain.name}/check/",
headers={"Authorization": f"Basic {self.API_CREDENTIALS}"},
verify=True,
timeout=10,
)
if response.status_code == status.HTTP_200_OK:
dimail_status = response.json()["state"]
if (
domain.status != enums.MailDomainStatusChoices.ENABLED
and dimail_status == "ok"
):
self.enable_pending_mailboxes(domain)
domain.status = enums.MailDomainStatusChoices.ENABLED
domain.save()
elif (
domain.status != enums.MailDomainStatusChoices.FAILED
and dimail_status == "broken"
):
domain.status = enums.MailDomainStatusChoices.FAILED
domain.save()
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."""
@@ -441,3 +415,82 @@ class DimailAPIClient:
recipient=mailbox.secondary_email,
mailbox_data=response.json(),
)
def check_domain(self, domain):
"""Send a request to dimail to check domain health."""
response = session.get(
f"{self.API_URL}/domains/{domain.name}/check/",
headers={"Authorization": f"Basic {self.API_CREDENTIALS}"},
verify=True,
timeout=10,
)
if response.status_code == status.HTTP_200_OK:
return response.json()
return self.raise_exception_for_unexpected_response(response)
def fix_domain(self, domain):
"""Send a request to dimail to fix a domain.
Allow to fix internal checks."""
response = session.get(
f"{self.API_URL}/domains/{domain.name}/fix/",
headers={"Authorization": f"Basic {self.API_CREDENTIALS}"},
verify=True,
timeout=10,
)
if response.status_code == status.HTTP_200_OK:
logger.info(
"Domain %s successfully fixed by dimail",
str(domain),
)
return response.json()
return self.raise_exception_for_unexpected_response(response)
def fetch_domain_status(self, domain):
"""Send a request to check and update status of a domain."""
dimail_response = self.check_domain(domain)
dimail_state = dimail_response["state"]
# if domain is not enabled and dimail returns ok status, enable it
if (
domain.status != enums.MailDomainStatusChoices.ENABLED
and dimail_state == "ok"
):
self.enable_pending_mailboxes(domain)
domain.status = enums.MailDomainStatusChoices.ENABLED
domain.save()
# if dimail returns broken status, we need to extract external and internal checks
# and manage the case where the domain has to be fixed by support
elif dimail_state == "broken":
external_checks = self._get_dimail_checks(dimail_response, internal=False)
internal_checks = self._get_dimail_checks(dimail_response, internal=True)
# manage the case where the domain has to be fixed by support
if not all(external_checks.values()):
domain.status = enums.MailDomainStatusChoices.ACTION_REQUIRED
domain.save()
# if all external checks are ok but not internal checks, we need to fix internal checks
elif all(external_checks.values()) and not all(internal_checks.values()):
# a call to fix endpoint is needed because all external checks are ok
dimail_response = self.fix_domain(domain)
# we need to check again if all internal and external checks are ok
external_checks = self._get_dimail_checks(
dimail_response, internal=False
)
internal_checks = self._get_dimail_checks(
dimail_response, internal=True
)
if all(external_checks.values()) and all(internal_checks.values()):
domain.status = enums.MailDomainStatusChoices.ENABLED
domain.save()
elif all(external_checks.values()) and not all(
internal_checks.values()
):
domain.status = enums.MailDomainStatusChoices.FAILED
domain.save()
return dimail_response
def _get_dimail_checks(self, dimail_response, internal: bool):
checks = {
key: value
for key, value in dimail_response.items()
if isinstance(value, dict) and value.get("internal") is internal
}
return {key: value.get("ok", False) for key, value in checks.items()}