import faker import pytest from ows_text_campaigns.assets.models import Asset from ows_text_campaigns.audiences.models import Audience, AudienceTextFan from ows_text_campaigns.campaigns.enums import MessageChannel from ows_text_campaigns.campaigns.exceptions import CampaignAudienceRequiredError from ows_text_campaigns.campaigns.handlers import ( GetCampaignPricingEstimationHandler, GetCampaignPricingEstimationRequest, ) from ows_text_campaigns.campaigns.handlers.get_campaign_pricing_estimation import ( ChannelCost, Cost, CountryCost, ) from ows_text_campaigns.campaigns.models import Campaign, CampaignStatus from ows_text_campaigns.twilio.models import TwilioMessagePricingByCountry from tests.unit.types import CreateModel, CreatePgModel class TestGetCampaignPricingEstimationHandler: @pytest.mark.db def test_get_campaign_pricing_estimate_with_share( self, handler: GetCampaignPricingEstimationHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: country_1 = "AA" country_2 = "BB" message_channel = MessageChannel.SMS min_price_1 = 10 max_price_1 = 15 min_price_2 = 22 max_price_2 = 35 reply_share = faker.pyfloat(min_value=0.0, max_value=1) audience = create_pg_model(Audience, name="Test Audience") campaign = create_model( Campaign, name="Test Campaign", status=CampaignStatus.IN_PROGRESS, audience_id=audience.id, ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_1, ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_1, ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_2, ) create_model( TwilioMessagePricingByCountry, country_iso2=country_1, message_type=message_channel, min_current_price=min_price_1, max_current_price=max_price_1, ) create_model( TwilioMessagePricingByCountry, country_iso2=country_2, message_type=message_channel, min_current_price=min_price_2, max_current_price=max_price_2, ) response = handler.handle( GetCampaignPricingEstimationRequest( identity_id=identity_id, campaign_id=campaign.id, reply_share=reply_share, ) ) assert response assert response.countries == [ CountryCost( name=country_1, channels=[ ChannelCost( name=message_channel, number_of_fans=2, inbound_cost=Cost( min=20 * reply_share, max=30 * reply_share, ).rounded(), outbound_cost=Cost(min=20, max=30).rounded(), ) ], ), CountryCost( name=country_2, channels=[ ChannelCost( name=message_channel, number_of_fans=1, inbound_cost=Cost( min=22 * reply_share, max=35 * reply_share ).rounded(), outbound_cost=Cost(min=22, max=35).rounded(), ) ], ), ] assert ( response.total_cost == Cost( min=20 + 20 * reply_share + 22 + 22 * reply_share, max=30 + 30 * reply_share + 35 + 35 * reply_share, ).rounded() ) @pytest.mark.db def test_get_campaign_pricing_estimate_missing_audience( self, handler: GetCampaignPricingEstimationHandler, create_model: CreateModel, identity_id: str, ) -> None: campaign = create_model( Campaign, name="Test Campaign", status=CampaignStatus.IN_PROGRESS, audience_id=None, ) with pytest.raises(CampaignAudienceRequiredError): handler.handle( GetCampaignPricingEstimationRequest( identity_id=identity_id, campaign_id=campaign.id, reply_share=0.5, ) ) @pytest.mark.db def test_get_campaign_pricing_estimate_with_share_mms_countries( self, handler: GetCampaignPricingEstimationHandler, create_model: CreateModel, create_pg_model: CreatePgModel, identity_id: str, faker: faker.Faker, ) -> None: country_1 = "CA" country_2 = "US" country_3 = "AA" message_channel = MessageChannel.SMS min_price_1 = 10 max_price_1 = 15 min_price_2 = 22 max_price_2 = 35 min_price_3 = 43 max_price_3 = 78 reply_share = 0 audience = create_pg_model(Audience, name="Test Audience") asset = create_model(Asset, key=faker.pystr()) campaign = create_model( Campaign, name="Test Campaign", status=CampaignStatus.IN_PROGRESS, audience_id=audience.id, assets=[asset], ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_1, ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_1, ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_2, ) create_model( AudienceTextFan, audience_id=audience.id, channel=message_channel, fan_country=country_3, ) create_model( TwilioMessagePricingByCountry, country_iso2=country_1, message_type=message_channel, min_current_price=min_price_1, max_current_price=max_price_1, ) create_model( TwilioMessagePricingByCountry, country_iso2=country_2, message_type=message_channel, min_current_price=min_price_2, max_current_price=max_price_2, ) create_model( TwilioMessagePricingByCountry, country_iso2=country_3, message_type=message_channel, min_current_price=min_price_3, max_current_price=max_price_3, ) response = handler.handle( GetCampaignPricingEstimationRequest( identity_id=identity_id, campaign_id=campaign.id, reply_share=reply_share, ) ) assert response assert response.countries == [ CountryCost( name=country_3, channels=[ ChannelCost( name=message_channel, number_of_fans=1, inbound_cost=Cost(min=0.0, max=0.0).rounded(), outbound_cost=Cost(min=43, max=78).rounded(), ) ], ), CountryCost( name=country_1, channels=[ ChannelCost( name=message_channel, number_of_fans=2, inbound_cost=Cost( min=0.0, max=0.0, ).rounded(), outbound_cost=Cost( min=handler.mms_pricing_cost.min * 2, max=handler.mms_pricing_cost.max * 2, ).rounded(), ) ], ), CountryCost( name=country_2, channels=[ ChannelCost( name=message_channel, number_of_fans=1, inbound_cost=Cost( min=0.0, max=0.0, ).rounded(), outbound_cost=Cost( min=handler.mms_pricing_cost.min, max=handler.mms_pricing_cost.max, ).rounded(), ) ], ), ] assert ( response.total_cost == Cost( min=handler.mms_pricing_cost.min * 2 + handler.mms_pricing_cost.min + 43, max=handler.mms_pricing_cost.max * 2 + handler.mms_pricing_cost.max + 78, ).rounded() )