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.handlers import UpdateAutomatedEmailContentHandler
from email_campaigns.automated.models import (
AutomatedEmail,
AutomatedEmailTrigger,
SourceCampaign,
)
from email_campaigns.emails.exceptions import (
EmailOptInButtonMissingError,
EmailPrivacyFooterMissingError,
)
from email_campaigns.emails.handlers import UpdateEmailContentRequest
from tests.unit.types import CreateModel
class TestUpdateAutomatedEmailContentHandler:
@pytest.mark.db
def test_update_content_without_source_campaigns_missing_privacy_footer(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
automated_email = create_model(AutomatedEmail)
with pytest.raises(EmailPrivacyFooterMissingError):
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html="no footer",
)
)
@pytest.mark.db
def test_update_content_without_source_campaigns_opt_in_missing_button(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
automated_email = create_model(AutomatedEmail, type=AutomatedEmailType.OPT_IN)
with pytest.raises(EmailOptInButtonMissingError):
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html='',
)
)
@pytest.mark.db
def test_update_content_with_source_campaigns_and_privacy_footer(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
html = ''
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html=html,
)
)
assert automated_email.html_content == html
@pytest.mark.db
def test_update_content_with_source_campaigns_missing_privacy_footer(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
with pytest.raises(EmailPrivacyFooterMissingError):
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html="no footer",
)
)
@pytest.mark.db
def test_update_content_opt_in_with_source_campaigns_missing_button(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
type=AutomatedEmailType.OPT_IN,
html_content=(
''
''
),
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
with pytest.raises(EmailOptInButtonMissingError):
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html='',
)
)
@pytest.mark.db
def test_update_content_opt_in_with_source_campaigns_and_button(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
) -> None:
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
type=AutomatedEmailType.OPT_IN,
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
html = (
''
''
)
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html=html,
)
)
assert automated_email.html_content == html
class TestUpdateAutomatedEmailContentHandlerCompressedContent:
@pytest.mark.db
def test_update_content_writes_compressed_content(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
) -> None:
stripo_client_mock.compress.return_value = ""
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
css_content="p{color:red}",
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html='',
css="p{color:blue}",
)
)
assert (
automated_email.html_content
== ''
)
assert automated_email.css_content == "p{color:blue}"
assert automated_email.compressed_content == ""
stripo_client_mock.compress.assert_called_once_with(
'',
"p{color:blue}",
)
@pytest.mark.db
def test_update_content_compress_uses_existing_field_when_request_omits_it(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
) -> None:
stripo_client_mock.compress.return_value = ""
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
html_content='old
',
css_content="p{color:red}",
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html='new
',
# css intentionally omitted — must reuse existing css
)
)
stripo_client_mock.compress.assert_called_once_with(
'new
',
"p{color:red}",
)
assert automated_email.compressed_content == ""
@pytest.mark.db
def test_update_content_propagates_stripo_failure(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
campaign_public_preview_cache_mock: mock.MagicMock,
) -> None:
campaign_public_preview_cache_mock.reset_mock()
stripo_client_mock.compress.side_effect = StripoClientError("boom")
source_campaign = create_model(SourceCampaign)
automated_email = create_model(
AutomatedEmail,
html_content='old
',
css_content="p{color:red}",
compressed_content="",
source_campaign_connections=[
AutomatedEmailTrigger(source_campaign_id=source_campaign.id)
],
)
with pytest.raises(StripoClientError):
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html='new
',
css="p{color:blue}",
)
)
# compress raised before assignment, so old value preserved
assert automated_email.compressed_content == ""
# preview_cache.delete only runs after successful save — its absence
# confirms save did not run either.
campaign_public_preview_cache_mock.delete.assert_not_called()
@pytest.mark.db
def test_update_content_skips_compress_when_not_automated(
self,
handler: UpdateAutomatedEmailContentHandler,
create_model: CreateModel,
identity_id: str,
stripo_client_mock: mock.MagicMock,
) -> None:
automated_email = create_model(
AutomatedEmail,
css_content="p{color:red}",
compressed_content=None,
)
html = 'draft
'
handler.handle(
UpdateEmailContentRequest(
identity_id=identity_id,
email_id=automated_email.id,
html=html,
css="p{color:blue}",
)
)
assert automated_email.html_content == html
assert automated_email.css_content == "p{color:blue}"
assert automated_email.compressed_content is None
stripo_client_mock.compress.assert_not_called()