🐛(dimail) ignore oxadmin mailboxes when importing mailboxes

oxadmin mailbox are technical mailboxes used by dimail. It should
not be imported when importing mailboxes from dimail.
This commit is contained in:
Marie PUPO JEAMMET
2026-01-12 16:51:33 +01:00
committed by Marie
parent bc1cbef168
commit d2ef9e0beb
3 changed files with 77 additions and 63 deletions

View File

@@ -8,6 +8,7 @@ and this project adheres to
## [Unreleased] ## [Unreleased]
- 🐛(dimail) ignore oxadmin when importing mailboxes from dimail #986
- ✨(aliases) delete all aliases in one call #1002 - ✨(aliases) delete all aliases in one call #1002
- ✨(aliases) fix deleting single aliases #1002 - ✨(aliases) fix deleting single aliases #1002
- 🔥(plugins) remove CommuneCreation plugin - 🔥(plugins) remove CommuneCreation plugin

View File

@@ -6,9 +6,6 @@ Unit tests for dimail client
import logging import logging
import re import re
from email.errors import HeaderParseError, NonASCIILocalPartDefect
from logging import Logger
from unittest import mock
import pytest import pytest
import responses import responses
@@ -68,20 +65,27 @@ def test_dimail_synchronization__already_sync(dimail_token_ok):
@responses.activate @responses.activate
@mock.patch.object(Logger, "warning") def test_dimail_synchronization__synchronize_mailboxes(caplog, dimail_token_ok): # pylint: disable=W0613, R0914
def test_dimail_synchronization__synchronize_mailboxes(mock_warning, dimail_token_ok):
"""A mailbox existing solely on dimail should be synchronized """A mailbox existing solely on dimail should be synchronized
upon calling sync function on its domain""" upon calling sync function on its domain"""
caplog.set_level(logging.INFO)
domain = factories.MailDomainEnabledFactory() domain = factories.MailDomainEnabledFactory()
assert not models.Mailbox.objects.exists()
existing_alias = factories.AliasFactory(domain=domain) existing_alias = factories.AliasFactory(domain=domain)
dimail_client = DimailAPIClient() dimail_client = DimailAPIClient()
# Ensure successful response using "responses": # Ensure successful response using "responses":
# token response in fixtures # successful token in fixtures
mailbox_valid = { mailbox_valid = {
"type": "mailbox",
"status": "ok",
"email": f"validmailbox@{domain.name}",
"givenName": "Michael",
"surName": "Roch",
"displayName": "Michael Roch",
}
mailbox_oxadmin = {
"type": "mailbox", "type": "mailbox",
"status": "broken", "status": "broken",
"email": f"oxadmin@{domain.name}", "email": f"oxadmin@{domain.name}",
@@ -113,7 +117,7 @@ def test_dimail_synchronization__synchronize_mailboxes(mock_warning, dimail_toke
"surName": "Vang", "surName": "Vang",
"displayName": "Jean Vang", "displayName": "Jean Vang",
} }
mailbox_existing_username = { mailbox_existing_alias = {
"type": "mailbox", "type": "mailbox",
"status": "broken", "status": "broken",
"email": f"{existing_alias.local_part}@{domain.name}", "email": f"{existing_alias.local_part}@{domain.name}",
@@ -126,10 +130,11 @@ def test_dimail_synchronization__synchronize_mailboxes(mock_warning, dimail_toke
re.compile(rf".*/domains/{domain.name}/mailboxes/"), re.compile(rf".*/domains/{domain.name}/mailboxes/"),
json=[ json=[
mailbox_valid, mailbox_valid,
mailbox_oxadmin,
mailbox_with_wrong_domain, mailbox_with_wrong_domain,
mailbox_with_invalid_domain, mailbox_with_invalid_domain,
mailbox_with_invalid_local_part, mailbox_with_invalid_local_part,
mailbox_existing_username, mailbox_existing_alias,
], ],
status=status.HTTP_200_OK, status=status.HTTP_200_OK,
content_type="application/json", content_type="application/json",
@@ -137,29 +142,25 @@ def test_dimail_synchronization__synchronize_mailboxes(mock_warning, dimail_toke
imported_mailboxes = dimail_client.import_mailboxes(domain) imported_mailboxes = dimail_client.import_mailboxes(domain)
# 3 imports failed: wrong domain, HeaderParseError, NonASCIILocalPartDefect # 4 imports failed: oxadmin, wrong domain, HeaderParseError, NonASCIILocalPartDefect
assert mock_warning.call_count == 3 assert len(caplog.records) == 6
log_messages = [record.message for record in caplog.records]
# first we try to import email with a wrong domain expected_messages = [
assert mock_warning.call_args_list[0][0] == ( f"Not importing OX technical address: oxadmin@{domain.name}",
"Import of email %s failed because of a wrong domain", f"Import of email {mailbox_with_wrong_domain['email']} failed because of a wrong domain",
mailbox_with_wrong_domain["email"], f"Import of email {mailbox_with_invalid_domain['email']} failed with error Invalid Domain",
) f"Import of email {mailbox_with_invalid_local_part['email']} failed with error local-part \
contains non-ASCII characters)",
f"{existing_alias.local_part} already used in an existing alias.",
]
for message in expected_messages:
assert message in log_messages
# then we try to import email with invalid domain
invalid_mailbox_log = mock_warning.call_args_list[1][0]
assert invalid_mailbox_log[1] == mailbox_with_invalid_domain["email"]
assert isinstance(invalid_mailbox_log[2], HeaderParseError)
# finally we try to import email with non ascii local part
non_ascii_mailbox_log = mock_warning.call_args_list[2][0]
assert non_ascii_mailbox_log[1] == mailbox_with_invalid_local_part["email"]
assert isinstance(non_ascii_mailbox_log[2], NonASCIILocalPartDefect)
mailbox = models.Mailbox.objects.get()
assert mailbox.local_part == "oxadmin"
assert mailbox.status == enums.MailboxStatusChoices.ENABLED
assert imported_mailboxes == [mailbox_valid["email"]] assert imported_mailboxes == [mailbox_valid["email"]]
mailbox = models.Mailbox.objects.get()
assert mailbox.local_part == "validmailbox"
assert mailbox.status == enums.MailboxStatusChoices.ENABLED
@responses.activate @responses.activate

View File

@@ -375,43 +375,55 @@ class DimailAPIClient:
return self._raise_exception_for_unexpected_response(response) return self._raise_exception_for_unexpected_response(response)
dimail_mailboxes = response.json() dimail_mailboxes = response.json()
known_mailboxes = models.Mailbox.objects.filter(domain=domain) people_mailboxes = models.Mailbox.objects.filter(domain=domain)
known_aliases = [
known_alias.local_part
for known_alias in models.Alias.objects.filter(domain=domain)
]
imported_mailboxes = [] imported_mailboxes = []
for dimail_mailbox in dimail_mailboxes: for dimail_mailbox in dimail_mailboxes:
if ( try:
dimail_mailbox["email"] address = Address(addr_spec=dimail_mailbox["email"])
not in [str(known_mailboxes) for known_mailboxes in known_mailboxes] except (HeaderParseError, NonASCIILocalPartDefect) as error:
and dimail_mailbox["email"].split("@")[0] not in known_aliases logger.warning(
): "Import of email %s failed with error %s",
try: dimail_mailbox["email"],
# sometimes dimail api returns email from another domain, error,
# so we decide to exclude this kind of email )
address = Address(addr_spec=dimail_mailbox["email"]) continue
if address.domain == domain.name:
# creates a mailbox on our end if address.username == "oxadmin":
mailbox = models.Mailbox.objects.create( logger.warning(
first_name=dimail_mailbox["givenName"], "Not importing OX technical address: %s", dimail_mailbox["email"]
last_name=dimail_mailbox["surName"], )
local_part=address.username, continue
domain=domain,
status=enums.MailboxStatusChoices.ENABLED, if address.username in [
password=make_password(None), # unusable password alias_.local_part
) for alias_ in models.Alias.objects.filter(domain=domain)
imported_mailboxes.append(str(mailbox)) ]:
else: logger.warning(
logger.warning( "%s already used in an existing alias.",
"Import of email %s failed because of a wrong domain", address.username,
dimail_mailbox["email"], )
) continue
except (HeaderParseError, NonASCIILocalPartDefect) as err:
if str(address) not in [
str(people_mailbox) for people_mailbox in people_mailboxes
]:
# sometimes dimail api returns email from another domain,
# so we decide to exclude this kind of email
if address.domain == domain.name:
# creates a mailbox on our end
mailbox = models.Mailbox.objects.create(
first_name=dimail_mailbox["givenName"],
last_name=dimail_mailbox["surName"],
local_part=address.username,
domain=domain,
status=enums.MailboxStatusChoices.ENABLED,
password=make_password(None), # unusable password
)
imported_mailboxes.append(str(mailbox))
else:
logger.warning( logger.warning(
"Import of email %s failed with error %s", "Import of email %s failed because of a wrong domain",
dimail_mailbox["email"], dimail_mailbox["email"],
err,
) )
return imported_mailboxes return imported_mailboxes