import json import traceback from collections.abc import Iterator from datetime import datetime import faker import httpx import pytest import respx from fansifter_common.adapters.sendgrid import SendGridClient from fansifter_common.adapters.sendgrid.exceptions import SendGridClientError from fansifter_common.adapters.sendgrid.models import Email, FanData, MergeTags from fansifter_common.core.enums import Brand from fansifter_common.email import ( EmailType, encode_email_to_local_part, encode_reply_to_local_part, ) from fansifter_common.legal_info.types import LegalEntity REPLY_TO_SECRET_KEY = "test-secret-key" @pytest.fixture def client() -> Iterator[SendGridClient]: with SendGridClient( sme_api_key="sme_api_key", awal_api_key="awal_api_key", orchard_api_key="orchard_api_key", unsubscribe_host="unsubscribe_host", preference_center_url="preference_center_url", ) as client: yield client @pytest.fixture def client_with_reply_to() -> Iterator[SendGridClient]: with SendGridClient( sme_api_key="sme_api_key", awal_api_key="awal_api_key", orchard_api_key="orchard_api_key", unsubscribe_host="unsubscribe_host", preference_center_url="preference_center_url", reply_to_header_enabled=True, reply_to_sme_domain="reply.sme.com", reply_to_orchard_domain="reply.theorchard.com", reply_to_awal_domain="reply.awal.com", reply_to_secret_key=REPLY_TO_SECRET_KEY, ) as client: yield client def test_request_error(client: SendGridClient, respx_mock: respx.MockRouter) -> None: headers = { "Authorization": client._api_keys[Brand.SME], } respx_mock.get( "https://api.sendgrid.com/v3/mail/test", headers=headers, ).mock(return_value=httpx.Response(status_code=500)) try: client.request("GET", "https://api.sendgrid.com/v3/mail/test") output = "" except Exception as exc: output = "".join(traceback.format_exception(exc)) assert client._api_keys[Brand.SME] not in output def test_send_email( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": False, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v1", ) def test_send_email_includes_first_name_merge_tag( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: fan_with_name = FanData( email=faker.email(), profile_token=None, country_iso2="us", legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), merge_tags=MergeTags(first_name="John"), ) fan_without_name = FanData( email=faker.email(), profile_token=None, country_iso2="us", legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) route = respx_mock.post("https://api.sendgrid.com/v3/mail/send").mock( return_value=httpx.Response(202) ) client.send_mail( email_id=faker.pystr(), email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=faker.pystr(), vendor_id=faker.pyint(), subject=faker.pystr(), from_email=Email(name=faker.pystr(), email=faker.email()), to_fans=[fan_with_name, fan_without_name], html_content=faker.pystr(), preview_text=None, version="v1", ) sent_payload = json.loads(route.calls.last.request.content) personalizations = sent_payload["personalizations"] assert personalizations[0]["substitutions"]["-firstName-"] == "John" assert personalizations[1]["substitutions"]["-firstName-"] == "Fan" def test_send_email_for_sending_test_emails_with_missed_profile_token_user( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=None, country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": {}, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": False, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v1", ) def test_send_email_for_internal_orchard_label( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = 7123 # Orchard internal label vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.THEORCHARD]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": False, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.THEORCHARD, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v1", ) def test_send_email_for_non_internal_orchard_label( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() opt_in_external = "{accountName} {brandName}" fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=opt_in_external, unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.THEORCHARD]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{vendor_name} The Orchard
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
', "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": False, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.THEORCHARD, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v1", ) def test_build_personalization_headers_with_profile_token_and_seedlist( client: SendGridClient, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) profile_token = faker.pystr() headers = client._build_personalization_headers( email_id=campaign_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v1", fan_profile_token=profile_token, is_seedlist=True, ) expected_headers = { "X-InboxMonster-CID": campaign_id, "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", } assert headers == expected_headers def test_build_personalization_headers_with_profile_token_no_seedlist( client: SendGridClient, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) profile_token = faker.pystr() headers = client._build_personalization_headers( email_id=campaign_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v1", fan_profile_token=profile_token, is_seedlist=False, ) expected_headers = { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", } assert headers == expected_headers def test_build_personalization_headers_no_profile_token_with_seedlist( client: SendGridClient, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() headers = client._build_personalization_headers( email_id=campaign_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v1", fan_profile_token=None, is_seedlist=True, ) expected_headers = { "X-InboxMonster-CID": campaign_id, } assert headers == expected_headers def test_build_personalization_headers_no_profile_token_no_seedlist( client: SendGridClient, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() headers = client._build_personalization_headers( email_id=campaign_id, email_type=EmailType.CAMPAIGN, automated_email_trigger_id=None, version="v1", fan_profile_token=None, is_seedlist=False, ) assert headers == {} def test_send_email_with_seedlist( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "X-InboxMonster-CID": campaign_id, "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": True, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, is_seedlist=True, version="v1", ) def test_send_email_with_seedlist_no_profile_token( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=None, country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "X-InboxMonster-CID": campaign_id, }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": True, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, is_seedlist=True, version="v1", ) def test_send_email_with_is_test_true_and_seedlist_true( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "X-InboxMonster-CID": campaign_id, "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": True, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, is_test=True, is_seedlist=True, version="v1", ) def test_get_privacy_block_for_preview( client: SendGridClient, faker: faker.Faker ) -> None: unsubscribe_text = faker.pystr() privacy_links_block = faker.pystr() vendor_name = faker.pystr() opt_in_external = "{accountName} {brandName}" unsubscribe_url = faker.pystr() footer = client.get_user_privacy_footer_block( email=faker.email(), privacy_links_block=privacy_links_block, opt_in_info=faker.pystr(), opt_in_info_external=opt_in_external, unsubscribe_text=unsubscribe_text, unsubscribe_url=unsubscribe_url, vendor_id=faker.pyint(), vendor_name=vendor_name, legal_entity=LegalEntity(name=faker.pystr(), address=faker.pystr()), brand=Brand.THEORCHARD, preview=True, ) assert footer == ( f'{vendor_name} The Orchard
' f"{privacy_links_block}
" f'{unsubscribe_text}
' ) def test_get_privacy_block_with_profile_link( client: SendGridClient, faker: faker.Faker ) -> None: unsubscribe_text = faker.pystr() privacy_links_block = faker.pystr() unsubscribe_url = faker.pystr() profile_text = faker.pystr() profile_url = faker.url() footer = client.get_user_privacy_footer_block( email=faker.email(), privacy_links_block=privacy_links_block, opt_in_info="{email}", opt_in_info_external=faker.pystr(), unsubscribe_text=unsubscribe_text, unsubscribe_url=unsubscribe_url, vendor_id=faker.pyint(), vendor_name=faker.pystr(), legal_entity=LegalEntity(name=faker.pystr(), address=faker.pystr()), brand=Brand.SME, profile_text=profile_text, profile_url=profile_url, ) assert ( f'{profile_text} | ' f'{unsubscribe_text}
' ) in footer def test_get_privacy_block_without_profile_text_shows_only_unsubscribe( client: SendGridClient, faker: faker.Faker ) -> None: unsubscribe_text = faker.pystr() unsubscribe_url = faker.pystr() footer = client.get_user_privacy_footer_block( email=faker.email(), privacy_links_block=faker.pystr(), opt_in_info="{email}", opt_in_info_external=faker.pystr(), unsubscribe_text=unsubscribe_text, unsubscribe_url=unsubscribe_url, vendor_id=faker.pyint(), vendor_name=faker.pystr(), legal_entity=LegalEntity(name=faker.pystr(), address=faker.pystr()), brand=Brand.SME, ) assert ( f'{unsubscribe_text}
' ) in footer def test_get_privacy_block_without_unsubscribe_url_omits_unsubscribe_line( client: SendGridClient, faker: faker.Faker ) -> None: privacy_links_block = faker.pystr() opt_in_info = "{email}" email = faker.email() legal_entity = LegalEntity(name=faker.pystr(), address=faker.pystr()) footer = client.get_user_privacy_footer_block( email=email, privacy_links_block=privacy_links_block, opt_in_info=opt_in_info, opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), unsubscribe_url=None, vendor_id=faker.pyint(), vendor_name=faker.pystr(), legal_entity=legal_entity, brand=Brand.SME, ) assert "unsubscribe" not in footer.lower() assert privacy_links_block in footer assert legal_entity.name in footer assert legal_entity.address in footer # Tests for encode_email_to_local_part and decode_local_part_to_email def test_build_personalization_headers_v2_invalid_email_type( client: SendGridClient, faker: faker.Faker, ) -> None: """Test that client catches ValueError from utils and reraises as SendGridClientError.""" with pytest.raises(SendGridClientError, match="Invalid email type"): client._build_personalization_headers( email_id=faker.pystr(), email_type="INVALID", # type: ignore automated_email_trigger_id=None, version="v2", fan_profile_token=faker.pystr(), is_seedlist=False, ) def test_send_email_v2_campaign( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Test send_mail with v2 format for campaign emails.""" campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(2) ] from_email = Email(name=faker.pystr(), email=faker.email()) # v2 format: base64(camp.campaign_id).v2 encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v2", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": False, }, }, ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v2", ) def test_send_email_v2_automated_without_trigger( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Test send_mail with v2 format for automated emails without automated_email_trigger_id.""" email_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(2) ] from_email = Email(name=faker.pystr(), email=faker.email()) # v2 format: base64(auto.email_id).v2 encoded_local_part = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, version="v2", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": "#", "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": email_id, "email_id": email_id, "email_type": "AUTOMATED", "automated_email_trigger_id": None, "is_test": False, }, }, ) client.send_mail( email_id=email_id, email_type=EmailType.AUTOMATED, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v2", ) def test_send_email_v2_automated_with_trigger( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Test send_mail with v2 format for automated emails with automated_email_trigger_id.""" email_id = faker.pystr() automated_email_trigger_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() preview_text = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() double_opt_in_url = faker.url() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), double_opt_in_url=double_opt_in_url, ) for _ in range(2) ] from_email = Email(name=faker.pystr(), email=faker.email()) # v2 format: base64(auto.email_id.automated_email_trigger_id).v2 encoded_local_part = encode_email_to_local_part( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=automated_email_trigger_id, version="v2", ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", headers={ "Authorization": f"Bearer {client._api_keys[Brand.SME]}", "Content-Type": "application/json", }, json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": preview_text, "-doubleOptInUrl-": double_opt_in_url, "-privacyFooterBlock-": f'{fan.opt_in_info.format(email=f"""{fan.email}""")}
' f"{fan.privacy_links_block}
" f'{fan.unsubscribe_text}
' f'© {datetime.now().year} {fan.legal_entity_name}
' f"{fan.legal_entity_address}", "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": {"name": from_email.name, "email": from_email.email}, "reply_to": {"name": from_email.name, "email": from_email.email}, "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": email_id, "email_id": email_id, "email_type": "AUTOMATED", "automated_email_trigger_id": automated_email_trigger_id, "is_test": False, }, }, ) client.send_mail( email_id=email_id, email_type=EmailType.AUTOMATED, automated_email_trigger_id=automated_email_trigger_id, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, preview_text=preview_text, version="v2", ) def test_send_email_with_reply_to_header( client_with_reply_to: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Reply-To header is set per-fan with signed encoded email.""" campaign_id = faker.uuid4() subject = faker.pystr() html_content = faker.pystr() from_email = Email(name=faker.pystr(), email=faker.email()) fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(2) ] encoded_local_part = encode_email_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, version="v1", ) encoded_reply_to = encode_reply_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, key=REPLY_TO_SECRET_KEY, ) respx_mock.post( "https://api.sendgrid.com/v3/mail/send", json={ "personalizations": [ { "to": [{"email": fan.email}], "substitutions": { "-previewText-": "", "-doubleOptInUrl-": "#", "-privacyFooterBlock-": client_with_reply_to.get_user_privacy_footer_block( email=fan.email, privacy_links_block=fan.privacy_links_block, opt_in_info=fan.opt_in_info, opt_in_info_external=fan.opt_in_info_external, unsubscribe_text=fan.unsubscribe_text, unsubscribe_url=f"preference_center_url/profile/{fan.profile_token}/subscriptions", vendor_id=1, vendor_name="vendor", legal_entity=LegalEntity( name=fan.legal_entity_name, address=fan.legal_entity_address, ), brand=Brand.SME, ), "-firstName-": "Fan", }, "headers": { "List-Unsubscribe-Post": "List-Unsubscribe=One-Click", "List-Unsubscribe": f", ", }, } for fan in fans ], "from": from_email.get(), "subject": subject, "content": [{"type": "text/html", "value": html_content}], "custom_args": { "campaign_id": campaign_id, "email_id": campaign_id, "email_type": "CAMPAIGN", "automated_email_trigger_id": None, "is_test": False, }, "reply_to": { "email": f"{encoded_reply_to}@reply.sme.com", "name": "Sony Music", }, }, ) client_with_reply_to.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name="vendor", vendor_id=1, subject=subject, from_email=from_email, to_fans=fans, html_content=html_content, version="v1", ) def test_send_email_reply_to_disabled( respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """No Reply-To header when reply_to_header_enabled=False.""" campaign_id = faker.uuid4() fan = FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) from_email = Email(name=faker.pystr(), email=faker.email()) with SendGridClient( sme_api_key="sme_api_key", awal_api_key="awal_api_key", orchard_api_key="orchard_api_key", unsubscribe_host="unsubscribe_host", preference_center_url="preference_center_url", reply_to_header_enabled=False, reply_to_sme_domain="reply.sme.com", reply_to_secret_key=REPLY_TO_SECRET_KEY, ) as client: route = respx_mock.post("https://api.sendgrid.com/v3/mail/send").mock( return_value=httpx.Response(200) ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name="vendor", vendor_id=1, subject=faker.pystr(), from_email=from_email, to_fans=[fan], html_content=faker.pystr(), version="v1", ) sent_payload = route.calls[0].request body = json.loads(sent_payload.content) assert body["reply_to"] == from_email.get() def test_send_email_reply_to_no_domain_for_brand( respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Falls back to from_email reply_to when domain is not configured for the brand.""" campaign_id = faker.uuid4() fan = FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) from_email = Email(name=faker.pystr(), email=faker.email()) with SendGridClient( sme_api_key="sme_api_key", awal_api_key="awal_api_key", orchard_api_key="orchard_api_key", unsubscribe_host="unsubscribe_host", preference_center_url="preference_center_url", reply_to_header_enabled=True, reply_to_sme_domain=None, # SME domain not configured reply_to_secret_key=REPLY_TO_SECRET_KEY, ) as client: route = respx_mock.post("https://api.sendgrid.com/v3/mail/send").mock( return_value=httpx.Response(200) ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name="vendor", vendor_id=1, subject=faker.pystr(), from_email=from_email, to_fans=[fan], html_content=faker.pystr(), version="v1", ) sent_payload = route.calls[0].request body = json.loads(sent_payload.content) assert body["reply_to"] == from_email.get() def test_reply_to_is_same_for_all_fans( client_with_reply_to: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """All fans share the same top-level reply_to (signed by email_id, not fan email).""" campaign_id = faker.uuid4() fans = [ FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text=faker.pystr(), ) for _ in range(3) ] from_email = Email(name=faker.pystr(), email=faker.email()) route = respx_mock.post("https://api.sendgrid.com/v3/mail/send").mock( return_value=httpx.Response(200) ) client_with_reply_to.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name="vendor", vendor_id=1, subject=faker.pystr(), from_email=from_email, to_fans=fans, html_content=faker.pystr(), version="v1", ) body = json.loads(route.calls[0].request.content) expected_local = encode_reply_to_local_part( email_id=campaign_id, email_type=EmailType.CAMPAIGN, key=REPLY_TO_SECRET_KEY, ) assert body["reply_to"] == { "email": f"{expected_local}@reply.sme.com", "name": "Sony Music", } def test_send_email_v2_campaign_with_profile_text( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Test send_mail renders profile link before unsubscribe when profile_text is set.""" campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() profile_text = "Deine Daten verwalten" fan = FanData( email=faker.email(), profile_token=faker.pystr(), country_iso2="de", legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info=faker.pystr(), opt_in_info_external=faker.pystr(), unsubscribe_text="Abmelden", profile_text=profile_text, ) from_email = Email(name=faker.pystr(), email=faker.email()) route = respx_mock.post("https://api.sendgrid.com/v3/mail/send").mock( return_value=httpx.Response(200) ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=[fan], html_content=html_content, version="v2", ) body = json.loads(route.calls[0].request.content) footer_html = body["personalizations"][0]["substitutions"]["-privacyFooterBlock-"] profile_url = f"preference_center_url/profile/{fan.profile_token}" assert f'href="{profile_url}">{profile_text} | ' in footer_html assert f'href="{profile_url}/subscriptions">Abmelden' in footer_html def test_send_email_no_profile_token_not_test_omits_unsubscribe( client: SendGridClient, respx_mock: respx.MockRouter, faker: faker.Faker, ) -> None: """Real fan without profile_id: footer has no unsubscribe, no List-Unsubscribe headers.""" campaign_id = faker.pystr() subject = faker.pystr() html_content = faker.pystr() vendor_id = faker.pyint() vendor_name = faker.pystr() fan = FanData( email=faker.email(), profile_token=None, country_iso2=faker.pystr(), legal_entity_name=faker.pystr(), legal_entity_address=faker.pystr(), privacy_links_block=faker.pystr(), opt_in_info="{email}", opt_in_info_external=faker.pystr(), unsubscribe_text="Unsubscribe", ) from_email = Email(name=faker.pystr(), email=faker.email()) route = respx_mock.post("https://api.sendgrid.com/v3/mail/send").mock( return_value=httpx.Response(200) ) client.send_mail( email_id=campaign_id, email_type=EmailType.CAMPAIGN, brand=Brand.SME, vendor_name=vendor_name, vendor_id=vendor_id, subject=subject, from_email=from_email, to_fans=[fan], html_content=html_content, version="v2", is_test=False, ) body = json.loads(route.calls[0].request.content) personalization = body["personalizations"][0] footer_html = personalization["substitutions"]["-privacyFooterBlock-"] headers = personalization["headers"] assert "Unsubscribe" not in footer_html assert 'href="#"' not in footer_html assert "List-Unsubscribe" not in headers assert "List-Unsubscribe-Post" not in headers assert fan.privacy_links_block in footer_html assert fan.legal_entity_name in footer_html