from unittest import mock
import pytest
from fansifter_common.adapters.stripo.exceptions import StripoClientError
from email_campaigns.automated.enums import AutomatedEmailType
from email_campaigns.automated.exceptions import (
SourceCampaignConnectionsEmptyError,
SourceCampaignSubscriptionOnlyRequiredError,
SourceCampaignVendorMismatchError,
)
from email_campaigns.automated.handlers import (
AttachSourceCampaignsHandler,
AttachSourceCampaignsRequest,
)
from email_campaigns.automated.models import (
AutomatedEmail,
AutomatedEmailTrigger,
SourceCampaign,
)
from email_campaigns.automated.types import SourceCampaignConnection
from email_campaigns.emails.exceptions import (
EmailContentRequiredFieldsError,
EmailOptInButtonMissingError,
EmailPrivacyFooterMissingError,
EmailSenderEmailNotValidError,
)
from tests.unit.types import CreateModel
class TestAttachSourceCampaignsHandler:
@pytest.mark.db
def test_attach_source_campaigns(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign_1 = create_model(SourceCampaign, vendor_id=None)
source_campaign_2 = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
html_content='',
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign_1.id)
],
)
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign_1.id),
SourceCampaignConnection(source_campaign_id=source_campaign_2.id),
],
)
)
assert len(automated_email.source_campaign_connections) == 2
@pytest.mark.db
def test_attach_source_campaigns_empty_connections(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
automated_email = create_model(AutomatedEmail)
with pytest.raises(SourceCampaignConnectionsEmptyError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[],
)
)
@pytest.mark.db
def test_attach_source_campaigns_missing_required_content(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(AutomatedEmail, sender_name=None)
with pytest.raises(EmailContentRequiredFieldsError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
@pytest.mark.db
def test_attach_source_campaigns_invalid_sender_email(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
email_username="not-a-valid-email-user@@",
html_content='',
)
with pytest.raises(EmailSenderEmailNotValidError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
@pytest.mark.db
def test_attach_source_campaigns_without_privacy_footer(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
html_content="",
)
with pytest.raises(EmailPrivacyFooterMissingError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
@pytest.mark.db
def test_attach_source_campaigns_opt_in_without_button(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
type=AutomatedEmailType.OPT_IN,
html_content=(''),
)
with pytest.raises(EmailOptInButtonMissingError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
@pytest.mark.db
def test_attach_source_campaigns_opt_in_with_button(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
type=AutomatedEmailType.OPT_IN,
html_content=(
""
''
''
""
),
)
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(
source_campaign_id=source_campaign.id,
subscription_only=True,
),
],
)
)
assert len(automated_email.source_campaign_connections) == 1
@pytest.mark.db
def test_attach_source_campaigns_updates_existing_connection(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
html_content='',
source_campaign_connections=[
AutomatedEmailTrigger(
source_campaign_id=source_campaign.id,
subscription_only=False,
)
],
)
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(
source_campaign_id=source_campaign.id,
subscription_only=True,
),
],
)
)
assert len(automated_email.source_campaign_connections) == 1
assert automated_email.source_campaign_connections[0].subscription_only is True
@pytest.mark.db
def test_attach_source_campaigns_vendor_mismatch(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
automated_email = create_model(
AutomatedEmail,
vendor_id=1,
html_content='',
)
source_campaign = create_model(SourceCampaign, vendor_id=2)
with pytest.raises(SourceCampaignVendorMismatchError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
@pytest.mark.db
def test_attach_source_campaigns_opt_in_without_subscription_only(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
type=AutomatedEmailType.OPT_IN,
html_content=(
""
''
''
""
),
)
with pytest.raises(SourceCampaignSubscriptionOnlyRequiredError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(
source_campaign_id=source_campaign.id,
subscription_only=False,
),
],
)
)
@pytest.mark.db
def test_attach_writes_compressed_content(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
) -> None:
stripo_client_mock.compress.return_value = ""
source_campaign = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
html_content='',
css_content="p{color:red}",
compressed_content=None,
)
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
assert automated_email.compressed_content == ""
stripo_client_mock.compress.assert_called_once_with(
'',
"p{color:red}",
)
@pytest.mark.db
def test_attach_recomputes_compressed_content_on_subsequent_attach(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
) -> None:
stripo_client_mock.compress.return_value = ""
source_campaign_1 = create_model(SourceCampaign, vendor_id=None)
source_campaign_2 = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
html_content='',
css_content="p{color:red}",
compressed_content="",
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign_1.id)
],
)
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign_1.id),
SourceCampaignConnection(source_campaign_id=source_campaign_2.id),
],
)
)
assert automated_email.compressed_content == ""
stripo_client_mock.compress.assert_called_once()
@pytest.mark.db
def test_attach_propagates_stripo_failure(
self,
handler: AttachSourceCampaignsHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
) -> None:
stripo_client_mock.compress.side_effect = StripoClientError("boom")
source_campaign = create_model(SourceCampaign, vendor_id=None)
automated_email = create_model(
AutomatedEmail,
html_content='',
css_content="p{color:red}",
compressed_content="",
)
with pytest.raises(StripoClientError):
handler.handle(
AttachSourceCampaignsRequest(
identity_id=identity_id,
email_id=automated_email.id,
source_campaign_connections=[
SourceCampaignConnection(source_campaign_id=source_campaign.id),
],
)
)
# compress raised before save — old compressed_content preserved in memory
assert automated_email.compressed_content == ""