🐛(activation-codes) create contact in brevo before add to list

Brevo requires the contact to exist before being added to a list.
This commit is contained in:
Quentin BEY
2025-10-21 14:18:25 +02:00
parent 6b2327b0f1
commit ca0c3a0169
5 changed files with 95 additions and 5 deletions
+1
View File
@@ -18,6 +18,7 @@ and this project adheres to
- 🐛(front) code activation fix session end #93
- 💬(wording) error page wording #102
- ⚡️(web-search) allow to override returned chunks #107
- 🐛(activation-codes) create contact in brevo before add to list #108
## [0.0.1] - 2025-10-19
@@ -293,6 +293,11 @@ def test_activation_code_use_success_notify_brevo(settings):
status=201,
)
brevo_create_contact = responses.post(
"https://api.brevo.com/v3/contacts",
status=200,
)
brevo_add_mock = responses.post(
"https://api.brevo.com/v3/contacts/lists/test_followup_list_name/contacts/add",
json={"message": "Contacts added successfully"},
@@ -311,6 +316,13 @@ def test_activation_code_use_success_notify_brevo(settings):
assert brevo_remove_mock.calls[0].request.headers["api-key"] == "test_brevo_api_key"
assert json.loads(brevo_remove_mock.calls[0].request.body) == {"emails": [user.email]}
assert len(brevo_create_contact.calls) == 1
assert brevo_create_contact.calls[0].request.headers["api-key"] == "test_brevo_api_key"
assert json.loads(brevo_create_contact.calls[0].request.body) == {
"email": user.email,
"updateEnabled": True,
}
assert len(brevo_add_mock.calls) == 1
assert brevo_add_mock.calls[0].request.headers["api-key"] == "test_brevo_api_key"
assert json.loads(brevo_add_mock.calls[0].request.body) == {"emails": [user.email]}
@@ -332,6 +332,11 @@ def test_register_email_success_brevo(api_client, settings):
settings.BREVO_API_KEY = "test_brevo_api_key"
settings.BREVO_WAITING_LIST_ID = "test_waiting_list_id"
brevo_create_contact = responses.post(
"https://api.brevo.com/v3/contacts",
status=200,
)
brevo_mock = responses.post(
"https://api.brevo.com/v3/contacts/lists/test_waiting_list_id/contacts/add",
json={"message": "Contacts added successfully"},
@@ -351,6 +356,13 @@ def test_register_email_success_brevo(api_client, settings):
registration = UserRegistrationRequest.objects.get(user=user)
assert registration.user == user
assert len(brevo_create_contact.calls) == 1
assert brevo_create_contact.calls[0].request.headers["api-key"] == "test_brevo_api_key"
assert json.loads(brevo_create_contact.calls[0].request.body) == {
"email": user.email,
"updateEnabled": True,
}
assert len(brevo_mock.calls) == 1
assert brevo_mock.calls[0].request.headers["api-key"] == "test_brevo_api_key"
assert json.loads(brevo_mock.calls[0].request.body) == {"emails": [user.email]}
@@ -373,6 +385,11 @@ def test_register_email_success_brevo_fails(api_client, settings):
settings.BREVO_API_KEY = "test_brevo_api_key"
settings.BREVO_WAITING_LIST_ID = "test_waiting_list_id"
_brevo_create_contact = responses.post(
"https://api.brevo.com/v3/contacts",
status=200,
)
brevo_mock = responses.post(
"https://api.brevo.com/v3/contacts/lists/test_waiting_list_id/contacts/add",
status=400,
+48
View File
@@ -10,6 +10,47 @@ import requests
logger = logging.getLogger(__name__)
def create_contact_in_brevo(email: str) -> bool:
"""
Create a contact in Brevo.
Args:
email (str): The email address of the user.
"""
api_key = settings.BREVO_API_KEY
if not api_key:
logger.info("Brevo API key not configured: skipping creating contact")
return False
url = "https://api.brevo.com/v3/contacts"
headers = {
"accept": "application/json",
"api-key": api_key,
"content-type": "application/json",
}
payload = {
"email": email,
"updateEnabled": True,
}
try:
response = requests.post(url, json=payload, headers=headers, timeout=5)
except requests.RequestException as e:
logger.exception(e)
return False
if not response.ok:
logger.error(
"Error creating contact in Brevo %s: (%s) %s",
email,
response.status_code,
response.text,
)
return False
return True
def add_user_to_brevo_list(emails: List[str], list_id: Optional[str]) -> None:
"""
Add email list to a Brevo list.
@@ -24,6 +65,13 @@ def add_user_to_brevo_list(emails: List[str], list_id: Optional[str]) -> None:
logger.info("Brevo API key or list ID not configured: skipping adding contact")
return
for email in emails:
# Ensure the contact exists before adding to the list
# `emails` contains several emails only when used from the admin interface bulk action
if not create_contact_in_brevo(email):
logger.error("Failed to create contact %s in Brevo, skipping adding to list", email)
return
url = f"https://api.brevo.com/v3/contacts/lists/{list_id}/contacts/add"
headers = {
"accept": "application/json",
@@ -507,7 +507,11 @@ def test_authentication_user_added_to_brevo(monkeypatch, rf, settings):
settings.BREVO_FOLLOWUP_LIST_ID = "follow-up-list-id"
settings.ACTIVATION_REQUIRED = False
brevo_mock = responses.post(
brevo_create_contact = responses.post(
"https://api.brevo.com/v3/contacts",
status=200,
)
brevo_add_to_list = responses.post(
"https://api.brevo.com/v3/contacts/lists/follow-up-list-id/contacts/add",
status=400,
)
@@ -545,9 +549,16 @@ def test_authentication_user_added_to_brevo(monkeypatch, rf, settings):
code_verifier="test-code-verifier",
)
assert len(brevo_mock.calls) == 1
assert brevo_mock.calls[0].request.headers["api-key"] == "test-api-key"
assert json.loads(brevo_mock.calls[0].request.body) == {"emails": [user.email]}
assert len(brevo_create_contact.calls) == 1
assert brevo_create_contact.calls[0].request.headers["api-key"] == "test-api-key"
assert json.loads(brevo_create_contact.calls[0].request.body) == {
"email": user.email,
"updateEnabled": True,
}
assert len(brevo_add_to_list.calls) == 1
assert brevo_add_to_list.calls[0].request.headers["api-key"] == "test-api-key"
assert json.loads(brevo_add_to_list.calls[0].request.body) == {"emails": [user.email]}
# Now test when activation is required: user should not be added to Brevo list
settings.ACTIVATION_REQUIRED = True
@@ -558,4 +569,5 @@ def test_authentication_user_added_to_brevo(monkeypatch, rf, settings):
code_verifier="test-code-verifier",
)
assert len(brevo_mock.calls) == 1 # No new call made
assert len(brevo_create_contact.calls) == 1 # No new call made
assert len(brevo_add_to_list.calls) == 1 # No new call made