from datetime import timedelta from typing import Any from unittest import mock import fakeredis import httpx import pytest from fansifter_common.adapters.twilio.exceptions import TwilioClientError from fansifter_common.adapters.twilio.models import SendMessageResponse from fansifter_common.utils import timezone from freezegun import freeze_time from app import lock from app.config import settings from app.enums import CampaignStatus from app.handler import handle from app.models import ( ArtistSettings, BatchRecipient, BatchSendRecord, Campaign, CampaignBatch, TwilioAccount, ) from app.types import PersonalizedAttributes, RenderedMessage from tests.unit.helpers import create_model, create_pg_model @pytest.mark.db class TestHandle: def test_completes_a_batch_in_one_chunk( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=3) for _ in range(3): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( lambda *, attributes, **_kwargs: [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ) twilio_client_mock.send_message.return_value = SendMessageResponse(sid="SM1") handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 3 assert updated_batch.completed_at is not None assert twilio_client_mock.send_message.call_count == 3 assert ( BatchSendRecord.query.where(BatchSendRecord.batch_id == batch.id).count() == 1 ) def test_flips_campaign_to_sent_only_once_every_batch_is_done( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, ) -> None: campaign = create_model(Campaign) batch_a = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=2) for _ in range(2): create_model(BatchRecipient, batch_id=batch_a.id) batch_b = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=2) for _ in range(2): create_model(BatchRecipient, batch_id=batch_b.id) lock.acquire(batch_a.id) lock.acquire(batch_b.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( lambda *, attributes, **_kwargs: [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ) twilio_client_mock.send_message.return_value = SendMessageResponse(sid="SM1") handle( { "batch_id": batch_a.id, "campaign_id": campaign.id, "messages_to_send": 2, "safe_until": None, } ) reloaded = Campaign.query.get(campaign.id) assert reloaded is not None assert reloaded.status == CampaignStatus.IN_PROGRESS handle( { "batch_id": batch_b.id, "campaign_id": campaign.id, "messages_to_send": 2, "safe_until": None, } ) reloaded = Campaign.query.get(campaign.id) assert reloaded is not None assert reloaded.status == CampaignStatus.SENT def test_checkpoints_offset_after_every_chunk_not_once_at_the_end( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(settings, "sender_chunk_size", 2) campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=5) for _ in range(5): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( lambda *, attributes, **_kwargs: [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ) twilio_client_mock.send_message.return_value = SendMessageResponse(sid="SM1") handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 5, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 5 assert updated_batch.completed_at is not None # 3 chunks: 2 + 2 + 1 -- one BatchSendRecord per chunk, not one for the whole invoke. records = BatchSendRecord.query.where( BatchSendRecord.batch_id == batch.id ).all() assert len(records) == 3 assert sorted(r.batch_offset for r in records) == [2, 4, 5] def test_stops_before_safe_until_buffer_with_partial_progress_preserved( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, redis_client: fakeredis.FakeRedis, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(settings, "sender_chunk_size", 2) monkeypatch.setattr(settings, "safe_until_buffer_seconds", 100) campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=4) for _ in range(4): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) twilio_client_mock.send_message.return_value = SendMessageResponse(sid="SM1") now = timezone.now() with freeze_time(now) as frozen: def _render_and_tick( *, attributes: list[PersonalizedAttributes], **_kwargs: object ) -> list[RenderedMessage]: frozen.tick(delta=timedelta(seconds=150)) return [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( _render_and_tick ) safe_until = now + timedelta(seconds=200) handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 4, "safe_until": safe_until, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None # Only the first chunk should have gone out: deadline = safe_until - 100s = now # + 100s, and the first chunk's render call ticks the clock forward by 150s, # which is past that deadline before the second chunk is ever fetched. assert updated_batch.batch_offset == 2 assert updated_batch.completed_at is None assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None ) def test_stops_at_own_time_budget_when_safe_until_is_none( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, redis_client: fakeredis.FakeRedis, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(settings, "sender_chunk_size", 2) monkeypatch.setattr(settings, "sender_time_budget_seconds", 100) campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=4) for _ in range(4): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) twilio_client_mock.send_message.return_value = SendMessageResponse(sid="SM1") now = timezone.now() with freeze_time(now) as frozen: def _render_and_tick( *, attributes: list[PersonalizedAttributes], **_kwargs: object ) -> list[RenderedMessage]: frozen.tick(delta=timedelta(seconds=150)) return [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( _render_and_tick ) handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 4, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 2 assert updated_batch.completed_at is None assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None ) def test_a_non_retryable_failed_message_does_not_block_the_rest_of_the_chunk( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=3) recipients = [create_model(BatchRecipient, batch_id=batch.id) for _ in range(3)] bad_fan_id = recipients[0].fan_id lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( lambda *, attributes, **_kwargs: [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ) def _send(request: Any, **_kwargs: object) -> SendMessageResponse: if request.to == next( r.fan_phone_number for r in recipients if r.fan_id == bad_fan_id ): # No response attached -- e.g. a connection error/timeout. Not safe # to retry, since Twilio may have already accepted the request. raise TwilioClientError("boom") return SendMessageResponse(sid="SM_OK") twilio_client_mock.send_message.side_effect = _send handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None # Offset advances past the whole chunk regardless of the one failure. assert updated_batch.batch_offset == 3 assert updated_batch.completed_at is not None record = BatchSendRecord.query.where(BatchSendRecord.batch_id == batch.id).one() assert record.sent_count == 2 # No response back -- not retried, since it may have already been accepted. # 1 attempt for the bad recipient, 1 attempt each for the other two. assert twilio_client_mock.send_message.call_count == 1 + 2 def test_a_retryable_failed_message_is_retried_and_does_not_block_the_chunk( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=3) recipients = [create_model(BatchRecipient, batch_id=batch.id) for _ in range(3)] bad_fan_id = recipients[0].fan_id lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( lambda *, attributes, **_kwargs: [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ) def _send(request: Any, **_kwargs: object) -> SendMessageResponse: if request.to == next( r.fan_phone_number for r in recipients if r.fan_id == bad_fan_id ): # Twilio itself responded with a 5xx -- safe to retry. response = httpx.Response( 503, request=httpx.Request("POST", "https://api.twilio.com") ) raise TwilioClientError("boom", response=response) return SendMessageResponse(sid="SM_OK") twilio_client_mock.send_message.side_effect = _send handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 3 record = BatchSendRecord.query.where(BatchSendRecord.batch_id == batch.id).one() assert record.sent_count == 2 # 1 initial attempt + `sender_message_max_retries` (default 1) retries for # the bad recipient, 1 attempt each for the other two. assert twilio_client_mock.send_message.call_count == 2 + 2 def test_invalid_twilio_credentials_are_not_retried_and_are_error_logged( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=2) for _ in range(2): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( lambda *, attributes, **_kwargs: [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ) def _send(*_args: Any, **_kwargs: object) -> SendMessageResponse: # Twilio rejects the API key/secret -- a definitive, permanent failure. response = httpx.Response( 401, request=httpx.Request("POST", "https://api.twilio.com") ) raise TwilioClientError("Authenticate", response=response) twilio_client_mock.send_message.side_effect = _send with caplog.at_level("ERROR"): handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 2, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 2 record = BatchSendRecord.query.where(BatchSendRecord.batch_id == batch.id).one() assert record.sent_count == 0 # A 401 is a permanent, non-retryable failure -- 1 attempt per recipient. assert twilio_client_mock.send_message.call_count == 2 assert ( sum( 1 for record in caplog.records if "Failed to send message" in record.message ) == 2 ) def test_releases_the_redis_lock_when_no_recipients_remain( self, redis_client: fakeredis.FakeRedis, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=3) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None ) def test_returns_early_and_releases_lock_when_batch_not_active( self, redis_client: fakeredis.FakeRedis, ) -> None: campaign = create_model(Campaign) batch = create_model( CampaignBatch, campaign_id=campaign.id, batch_size=3, batch_offset=3, ) lock.acquire(batch.id) handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None ) def test_stops_when_campaign_is_cancelled_mid_flight( self, ows_text_campaigns_client_mock: mock.MagicMock, twilio_client_mock: mock.MagicMock, redis_client: fakeredis.FakeRedis, monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(settings, "sender_chunk_size", 2) campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=4) for _ in range(4): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) twilio_client_mock.send_message.return_value = SendMessageResponse(sid="SM1") def _render_and_cancel( *, attributes: list[PersonalizedAttributes], **_kwargs: object ) -> list[RenderedMessage]: campaign.status = CampaignStatus.CANCELLED campaign.save() return [ RenderedMessage( fan_id=attrs["fan_id"], channel=attrs["channel"], recipient=attrs["phone_number"], sender="+15550001111", message="hi", ) for attrs in attributes ] ows_text_campaigns_client_mock.render_messages_batch.side_effect = ( _render_and_cancel ) handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 4, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 2 assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None ) def test_aborts_without_sending_when_artist_has_no_twilio_account( self, twilio_client_mock: mock.MagicMock, redis_client: fakeredis.FakeRedis, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=3) for _ in range(3): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 0 twilio_client_mock.send_message.assert_not_called() assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None ) def test_aborts_without_sending_when_credentials_resolution_fails( self, twilio_client_mock: mock.MagicMock, kms_client_mock: mock.MagicMock, redis_client: fakeredis.FakeRedis, ) -> None: campaign = create_model(Campaign) batch = create_model(CampaignBatch, campaign_id=campaign.id, batch_size=3) for _ in range(3): create_model(BatchRecipient, batch_id=batch.id) lock.acquire(batch.id) create_model( ArtistSettings, global_participant_id=campaign.global_participant_id, twilio_account_sid="AC_TEST", twilio_messaging_service_sid="MG_TEST", ) create_pg_model( TwilioAccount, account_sid="AC_TEST", api_key_id="SK_TEST", api_secret="secret", ) kms_client_mock.decrypt.side_effect = RuntimeError("boom") handle( { "batch_id": batch.id, "campaign_id": campaign.id, "messages_to_send": 3, "safe_until": None, } ) updated_batch = CampaignBatch.query.get(batch.id) assert updated_batch is not None assert updated_batch.batch_offset == 0 twilio_client_mock.send_message.assert_not_called() assert ( redis_client.get(settings.redis_sender_lock_key.format(batch_id=batch.id)) is None )