import datetime import faker import freezegun import pytest from fansifter_common.utils import timezone from ows_text_campaigns.artist.exceptions import ArtistNotConfiguredError from ows_text_campaigns.artist.models import ArtistPhoneNumber, ArtistSettings from ows_text_campaigns.audiences.enums import AudienceTarget, FanDataListType from ows_text_campaigns.audiences.exceptions import InvalidAudienceIdError from ows_text_campaigns.audiences.models import ( Audience, AudienceFanDataList, AudienceTextFan, ) from ows_text_campaigns.campaigns.enums import CampaignStatus, MessageChannel from ows_text_campaigns.campaigns.exceptions import ( CampaignAlreadyQueuedError, CampaignMustBeDraftError, CampaignMustBeScheduledError, CampaignSendAtMustBeFutureError, ChannelNotAllowedInCountryError, EmptyCampaignContentError, ) from ows_text_campaigns.campaigns.handlers import ( ScheduleCampaignHandler, ScheduleCampaignRequest, ) from ows_text_campaigns.campaigns.models import Campaign from ows_text_campaigns.campaigns.types import TimezoneAdjustment from ows_text_campaigns.twilio.models import TwilioAccount from tests.unit.types import CreateModel, CreatePgModel class TestScheduleCampaignHandler: @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_schedule_queue_not_enabled( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() twilio_account = create_pg_model(TwilioAccount) create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], twilio_account_sid=twilio_account.account_sid, ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) send_at = timezone.now() + datetime.timedelta(minutes=11) send_at_timezone = "UTC" timezone_adjustments = [ TimezoneAdjustment.model_construct( country_code="US", iana_tz_id="Europe/London", ), ] handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=send_at, timezone=send_at_timezone, adjustments=timezone_adjustments, ) ) assert campaign.status == CampaignStatus.SCHEDULED assert campaign.send_at == send_at assert campaign.send_at_timezone == send_at_timezone assert campaign.timezone_adjustments == timezone_adjustments assert campaign.recipients_count == audience.fan_count @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_schedule_created( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() twilio_account = create_pg_model(TwilioAccount) create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], twilio_account_sid=twilio_account.account_sid, ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) send_at = timezone.now() + datetime.timedelta(minutes=10) send_at_timezone = "UTC" timezone_adjustments = [ TimezoneAdjustment.model_construct( country_code="US", iana_tz_id="Europe/London", ), ] handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=send_at, timezone=send_at_timezone, adjustments=timezone_adjustments, ) ) assert campaign.status == CampaignStatus.SCHEDULED assert campaign.send_at == send_at assert campaign.send_at_timezone == send_at_timezone assert campaign.timezone_adjustments == timezone_adjustments assert campaign.recipients_count == audience.fan_count @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_empty_content( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, content="", ) with pytest.raises(EmptyCampaignContentError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.future_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, ) ) @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_artist_missing_phone_numbers( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[], ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) with pytest.raises(ArtistNotConfiguredError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.future_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, ) ) @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_whatsapp_not_allowed_in_us( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() twilio_account = create_pg_model(TwilioAccount) create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], twilio_account_sid=twilio_account.account_sid, ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, channel=MessageChannel.WHATSAPP, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) create_model( AudienceTextFan, audience_id=audience.id, channel=MessageChannel.WHATSAPP, fan_country="US", ) with pytest.raises(ChannelNotAllowedInCountryError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=timezone.now() + datetime.timedelta(minutes=11), timezone="UTC", adjustments=None, ) ) assert campaign.status == CampaignStatus.DRAFT @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_whatsapp_allowed_outside_us( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() twilio_account = create_pg_model(TwilioAccount) create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], twilio_account_sid=twilio_account.account_sid, ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, channel=MessageChannel.WHATSAPP, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) create_model( AudienceTextFan, audience_id=audience.id, channel=MessageChannel.WHATSAPP, fan_country="GB", ) handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=timezone.now() + datetime.timedelta(minutes=11), timezone="UTC", adjustments=None, ) ) assert campaign.status == CampaignStatus.SCHEDULED @pytest.mark.db def test_schedule_campaign_is_not_draft( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: audience = create_pg_model(Audience) campaign = create_model( Campaign, status=CampaignStatus.IN_PROGRESS, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) with pytest.raises(CampaignMustBeDraftError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.future_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, ) ) @pytest.mark.db def test_schedule_campaign_send_at_must_be_future( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], ) audience = create_pg_model(Audience, target=AudienceTarget.TEXT) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) with pytest.raises(CampaignSendAtMustBeFutureError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.past_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, ) ) @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_schedule_campaign_missing_audience( self, handler: ScheduleCampaignHandler, create_model: CreateModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], ) campaign = create_model( Campaign, status=CampaignStatus.DRAFT, global_participant_id=global_participant_id, audience_id=None, ) with pytest.raises(InvalidAudienceIdError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.future_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, ) ) @pytest.mark.db @freezegun.freeze_time("2024-10-15 10:30:00") def test_edit_campaign_schedule( self, handler: ScheduleCampaignHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: global_participant_id = faker.uuid4() twilio_account = create_pg_model(TwilioAccount) create_model( ArtistSettings, global_participant_id=global_participant_id, phone_numbers=[ ArtistPhoneNumber( global_participant_id=global_participant_id, phone_number="+1234567890", ) ], twilio_account_sid=twilio_account.account_sid, ) audience = create_pg_model( Audience, target=AudienceTarget.TEXT, fan_count=100, fandata_lists=[ AudienceFanDataList( type=FanDataListType.ARTIST, list_id=global_participant_id, ) ], ) campaign = create_model( Campaign, status=CampaignStatus.SCHEDULED, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, audience_id=audience.id, ) send_at = faker.future_datetime(tzinfo=timezone.UTC) send_at_timezone = "UTC" timezone_adjustments = [ TimezoneAdjustment.model_construct( country_code="US", iana_tz_id="Europe/London", ), ] handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=send_at, timezone=send_at_timezone, adjustments=timezone_adjustments, edit_mode=True, ) ) assert campaign.status == CampaignStatus.SCHEDULED assert campaign.send_at == send_at assert campaign.send_at_timezone == send_at_timezone assert campaign.timezone_adjustments == timezone_adjustments assert campaign.recipients_count == audience.fan_count @pytest.mark.db def test_schedule_campaign_edit_not_scheduled( self, handler: ScheduleCampaignHandler, create_model: CreateModel, identity_id: str, faker: faker.Faker, ) -> None: campaign = create_model( Campaign, status=CampaignStatus.DRAFT, ) with pytest.raises(CampaignMustBeScheduledError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.future_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, edit_mode=True, ) ) @pytest.mark.db def test_schedule_campaign_edit_not_in_scheduled_queue( self, handler: ScheduleCampaignHandler, create_model: CreateModel, identity_id: str, faker: faker.Faker, ) -> None: campaign = create_model( Campaign, status=CampaignStatus.SCHEDULED, send_at=timezone.now(), ) with pytest.raises(CampaignAlreadyQueuedError): handler.handle( ScheduleCampaignRequest( identity_id=identity_id, campaign_id=campaign.id, send_at=faker.future_datetime(tzinfo=timezone.UTC), timezone="UTC", adjustments=None, edit_mode=True, ) )