diff --git a/src/backend/mailbox_manager/tests/test_utils_dimail_client.py b/src/backend/mailbox_manager/tests/test_utils_dimail_client.py index fb7ece60f..a76e4b2c4 100644 --- a/src/backend/mailbox_manager/tests/test_utils_dimail_client.py +++ b/src/backend/mailbox_manager/tests/test_utils_dimail_client.py @@ -3,6 +3,8 @@ """ import re +from logging import Logger +from unittest import mock import pytest import responses @@ -61,7 +63,8 @@ def test_dimail_synchronization__already_sync(): assert set(models.Mailbox.objects.filter(domain=domain)) == set(pre_sync_mailboxes) -def test_dimail_synchronization__synchronize_mailboxes(): +@mock.patch.object(Logger, "warning") +def test_dimail_synchronization__synchronize_mailboxes(mock_warning): """A mailbox existing solely on dimail should be synchronized upon calling sync function on its domain""" domain = factories.MailDomainEnabledFactory() @@ -77,33 +80,50 @@ def test_dimail_synchronization__synchronize_mailboxes(): status=status.HTTP_200_OK, content_type="application/json", ) + + mailbox = { + "type": "mailbox", + "status": "broken", + "email": f"oxadmin@{domain.name}", + "givenName": "Admin", + "surName": "Context", + "displayName": "Context Admin", + } + mailbox_with_wrong_domain = { + "type": "mailbox", + "status": "broken", + "email": f"johndoe@wrongdomain.com", + "givenName": "John", + "surName": "Doe", + "displayName": "John Doe", + } + mailbox_with_wrong_local_part = ( + { + "type": "mailbox", + "status": "broken", + "email": f"nawaké@{domain.name}", + "givenName": "Joe", + "surName": "Doe", + "displayName": "Joe Doe", + }, + ) + rsps.add( rsps.GET, re.compile(rf".*/domains/{domain.name}/mailboxes/"), body=str( - [ - { - "type": "mailbox", - "status": "broken", - "email": f"oxadmin@{domain.name}", - "givenName": "Admin", - "surName": "Context", - "displayName": "Context Admin", - }, - { - "type": "mailbox", - "status": "broken", - "email": f"johndoe@wrongdomain.com", - "givenName": "John", - "surName": "Doe", - "displayName": "John Doe", - }, - ] + [mailbox, mailbox_with_wrong_domain, mailbox_with_wrong_local_part] ), status=status.HTTP_200_OK, content_type="application/json", ) + imported_mailboxes = dimail_client.synchronize_mailboxes_from_dimail(domain) + assert mock_warning.call_count == 1 + assert mock_warning.call_args_list == ( + "Import of email %s failed", + f"nawaké@{domain.name}", + ) mailbox = models.Mailbox.objects.get() assert mailbox.local_part == "oxadmin" diff --git a/src/backend/mailbox_manager/utils/dimail.py b/src/backend/mailbox_manager/utils/dimail.py index 55a6c084d..363258ed4 100644 --- a/src/backend/mailbox_manager/utils/dimail.py +++ b/src/backend/mailbox_manager/utils/dimail.py @@ -2,6 +2,7 @@ import ast import smtplib +from email.headerregistry import Address from logging import getLogger from django.conf import settings @@ -200,20 +201,28 @@ def synchronize_mailboxes_from_dimail(self, domain): if not dimail_mailbox["email"] 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 - dimail_local_part, dimail_domain = dimail_mailbox["email"].split("@") - if dimail_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=dimail_local_part, - domain=domain, - secondary_email=dimail_mailbox[ - "email" - ], # secondary email is mandatory. Unfortunately, dimail doesn't store any. - # We temporarily give current email as secondary email. + try: + # sometimes dimail api returns email from another domain, so we decide to exclude + # this kind of email + dimail_local_part, dimail_domain = dimail_mailbox["email"].split( + "@" ) - imported_mailboxes.append(str(mailbox)) + address = Address( + dimail_mailbox["displayName"], dimail_local_part, dimail_domain + ) + 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, + secondary_email=dimail_mailbox[ + "email" + ], # secondary email is mandatory. Unfortunately, dimail doesn't store any. + # We temporarily give current email as secondary email. + ) + imported_mailboxes.append(str(mailbox)) + except Exception as err: + logger.warning("Import of email %s failed", dimail_mailbox["email"]) return imported_mailboxes