import datetime from datetime import timedelta import pytest from dirty_equals import IsList, IsStr from faker import Faker from fansifter_common.auth.identity import Identity from fansifter_common.utils import timezone from dmp.adapters.features import AUDIENCE_SHOW_ADD_SECONDARY_FANS_TOGGLE from dmp.audiences.dtos import AudienceCriteria from dmp.audiences.enums import ( AudienceExportReason, AudienceSharePlatform, AudienceTarget, ) from dmp.audiences.filters import ( AudienceFilters, BirthMonthInFilter, CityFilter, CrmCampaignFilter, DigitalServiceProviderInFilter, EmailCampaignAction, EmailCampaignsFilter, FanSegmentInFilter, HeavyRotationV2InFilter, LastEngagementFilter, Operator, RegionFilter, SongwhipCampaignInFilter, SongwhipPresavePageInFilter, ) from dmp.audiences.models import AudienceFan from dmp.audiences.repositories import AudienceFanRepository from dmp.audiences.types import FanDataListId from dmp.crm_campaigns.models import CrmCampaignIdFanDbt from dmp.email_campaigns.models import EmailAnalyticsByCampaignFanDbt from dmp.fandata.enums import ( BirthMonth, DigitalServiceProvider, FanSegment, HeavyRotation, HeavyRotationFilter, HeavyRotationTmp, LastEngagement, ) from dmp.fandata.models import ( AudienceFiltersAdCampaignDbt, AudienceFiltersCustomListAdCampaignDbt, AudienceFiltersCustomListEmailCampaignDbt, AudienceFiltersEmailCampaignDbt, AudienceFiltersTextCampaignDbt, FanLocationDbt, FanPersonalDataDbt, GlobalAudienceFiltersAdCampaignDbt, TwilioFanSubscriptionStatus, ) from dmp.fandata.utils import make_fan_id from dmp.rosters.models import ArtistRosterLocalRep, ArtistRosterMainRep from dmp.songwhip.models import SongwhipCustomPageFanDbt, SongwhipPresavePageFanDbt from dmp.text_campaigns.models import ArtistPhoneNumber from tests.unit.faker import FakerTyped from tests.unit.types import ( CreateReportingModel, CreateReportingModelBatch, EnableFeatures, ) class TestAudienceFanRepository: # -- ADS target size -- @pytest.mark.db def test_get_size_empty( self, repository: AudienceFanRepository, fake: FakerTyped, identity: Identity, ) -> None: size = repository.get_size( AudienceCriteria( vendor_id=fake.integer(), subaccount_id=0, fandata_list_ids=[ FanDataListId.from_artist_id(fake.uuid4_string()), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert size == 0 @pytest.mark.db def test_get_size( self, repository: AudienceFanRepository, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: size = 3 global_participant_id = fake.uuid4_string() vendor_id = fake.integer() subaccount_id = fake.integer() create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, is_primary_fan=False, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert result == size @pytest.mark.db def test_get_size_internal_employee( self, repository: AudienceFanRepository, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, fake: FakerTyped, internal_employee_identity: Identity, ) -> None: size = 3 global_participant_id = fake.uuid4_string() vendor_id = fake.integer() subaccount_id = fake.integer() create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, is_primary_fan=False, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=internal_employee_identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert result == 4 @pytest.mark.db def test_get_size_with_target_ads( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) size = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert size == 2 @pytest.mark.db def test_get_size_with_countries( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, create_reporting_model_batch: CreateReportingModelBatch, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() country_1 = "US" country_2 = "UK" create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=2, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_1, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_2, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(countries=[country_1]), target=AudienceTarget.ADS, ) ) assert result == 2 @pytest.mark.db def test_get_size_with_excluded_countries( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, create_reporting_model_batch: CreateReportingModelBatch, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() country_1 = "US" country_2 = "UK" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_1, ) create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=2, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_2, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, excluded_countries=[country_1], ) ) assert result == 2 @pytest.mark.db def test_get_size_with_countries_and_excluded_countries( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, create_reporting_model_batch: CreateReportingModelBatch, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() country_1 = "US" country_2 = "EE" country_3 = "UK" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_1, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_2, ) create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=4, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_3, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( countries=[country_1, country_2, country_3] ), target=AudienceTarget.ADS, excluded_countries=[country_1, country_2], ) ) assert result == 4 @pytest.mark.db def test_get_size_with_ads_excluded_and_benelux_countries_not_excluded( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, fake: FakerTyped, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() country_nl = "NL" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_nl, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert result == 1 @pytest.mark.db def test_get_size_with_cities( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() city_id_1 = "1" city_id_2 = "2" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, city_id=city_id_1, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, city_id=city_id_2, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( cities=[ CityFilter.model_construct(place_id="1", city_id=city_id_1), CityFilter.model_construct(place_id="2", city_id=city_id_2), ] ), target=AudienceTarget.ADS, ) ) assert result == 2 @pytest.mark.db def test_get_size_with_cities_and_radius( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() city_id_1 = "1" city_id_2 = "2" # Tallinn create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, city_id=city_id_1, city_coordinates={"latitude": 59.4370, "longitude": 24.7536}, ) # Helsinki create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, city_id=city_id_2, city_coordinates={"latitude": 60.1695, "longitude": 24.9354}, ) criteria = AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( cities=[ CityFilter.model_construct( place_id="1", city_id=city_id_1, bounding_radius_in_km=13, longitude=24.7536, latitude=59.4370, ) ], radius_unit="KM", ), target=AudienceTarget.ADS, ) # Test with radius 13 km + 1 km, should include Tallinn and not include Helsinki criteria.filters.radius = 1 assert repository.get_size(criteria) == 1 # Test with radius 13 km + 100km, should include Tallinn and Helsinki criteria.filters.radius = 100 assert repository.get_size(criteria) == 2 @pytest.mark.db def test_get_size_with_cities_and_radius_in_mi( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() city_id_1 = "1" city_id_2 = "2" # New York create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, city_id=city_id_1, city_coordinates={"latitude": 40.7306, "longitude": -73.9352}, ) # Scranton create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, city_id=city_id_2, city_coordinates={"latitude": 41.4118, "longitude": -75.66524}, ) criteria = AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( cities=[ CityFilter.model_construct( place_id="1", city_id=city_id_1, longitude=-73.9352, latitude=40.7306, bounding_radius_in_km=24.5, # 15.22 mi ) ], radius_unit="MI", ), target=AudienceTarget.ADS, ) # Test with radius 15.22 mi + 1 mi, should include New York and not include Scranton criteria.filters.radius = 1 assert repository.get_size(criteria) == 1 # Test with radius 15.22 mi + 90 mi, should include New York and Scranton criteria.filters.radius = 90 assert repository.get_size(criteria) == 2 @pytest.mark.parametrize( "operator, values, expected", [ (Operator.INCLUDE, [HeavyRotation.IN], 1), (Operator.INCLUDE, [HeavyRotation.NOT_IN], 2), (Operator.INCLUDE, [HeavyRotation.IN, HeavyRotation.NOT_IN], 3), (Operator.EXCLUDE, [HeavyRotation.IN], 4), (Operator.EXCLUDE, [HeavyRotation.NOT_IN], 3), (Operator.EXCLUDE, [HeavyRotation.IN, HeavyRotation.NOT_IN], 2), (None, None, 5), ], ) @pytest.mark.db def test_get_size_with_heavy_rotation( self, operator: Operator | None, values: list[HeavyRotationFilter] | None, expected: int, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, create_reporting_model_batch: CreateReportingModelBatch, identity: Identity, fake: FakerTyped, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, heavy_rotation=HeavyRotationTmp.IN, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, heavy_rotation=HeavyRotationTmp.IN_BEFORE, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, heavy_rotation=HeavyRotationTmp.NOT_IN, ) create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=2, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, heavy_rotation=HeavyRotationTmp.UNKNOWN, ) result = repository.get_size( AudienceCriteria.model_construct( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( heavy_rotation_v2=HeavyRotationV2InFilter( operator=operator, value=values, ) if operator and values is not None else None ), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("spend_min", "spend_max", "expected"), [ ( 100.10, 150.20, 1, ), ( 100.10, 200.20, 2, ), ( 300.00, 500.00, 0, ), ], ) @pytest.mark.db def test_get_size_with_shopify_spend( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, spend_min: float, spend_max: float, expected: int, ) -> None: vendor_id = 1234 subaccount_id = 56789 global_participant_id = fake.uuid4_string() total_spend_1 = 100.10 total_spend_2 = 200.20 vendor_id_2 = vendor_id + 1 subaccount_id_2 = subaccount_id + 1 create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, total_spend=total_spend_1, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, total_spend=total_spend_2, ) # creating 3rd event for the same gp_id, account without shopify spend data fan_artist_3 = create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, total_spend=None, ) # creating event for different account for the same fan with shopify spend data create_reporting_model( AudienceFiltersAdCampaignDbt, fan_id=fan_artist_3.fan_id, vendor_id=vendor_id_2, subaccount_id=subaccount_id_2, global_participant_id=global_participant_id, total_spend=total_spend_2, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( spend_min=spend_min, spend_max=spend_max, ), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("dsp_1", "dsp_2", "dsp_3", "value", "operator", "expected"), [ ( DigitalServiceProvider.DEEZER, DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.YOUTUBE, [DigitalServiceProvider.YOUTUBE], Operator.INCLUDE, 2, ), ( DigitalServiceProvider.DEEZER, DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.YOUTUBE, [DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.DEEZER], Operator.INCLUDE, 3, ), ( DigitalServiceProvider.DEEZER, DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.YOUTUBE, [DigitalServiceProvider.SPOTIFY], Operator.INCLUDE, 0, ), ( DigitalServiceProvider.DEEZER, DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.YOUTUBE, [DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.DEEZER], Operator.EXCLUDE, 1, ), ( DigitalServiceProvider.DEEZER, DigitalServiceProvider.YOUTUBE, DigitalServiceProvider.AMAZON, [DigitalServiceProvider.AMAZON], Operator.INCLUDE, 1, ), ], ) @pytest.mark.db def test_get_size_with_dsp( self, dsp_1: DigitalServiceProvider, dsp_2: DigitalServiceProvider, dsp_3: DigitalServiceProvider, value: list[DigitalServiceProvider], expected: int, operator: Operator, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() country = "AA" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, dsp=[dsp_1, dsp_2], country_iso2=country, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, dsp=[dsp_3], country_iso2=country, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, dsp=[dsp_1], country_iso2=country, ) result = repository.get_size( AudienceCriteria.model_construct( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( dsp=DigitalServiceProviderInFilter(value=value, operator=operator) ), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("timestamp", "engagement", "operator", "expected"), [ ( timezone.now(), LastEngagement.PAST_2_WEEKS, Operator.INCLUDE, 1, ), ( timezone.now(), LastEngagement.PAST_2_WEEKS, Operator.EXCLUDE, 0, ), ( timezone.now() - timedelta(days=14), LastEngagement.PAST_2_WEEKS, Operator.INCLUDE, 1, ), ( timezone.now() - timedelta(days=15), LastEngagement.PAST_2_WEEKS, Operator.INCLUDE, 0, ), ( timezone.now() - timedelta(days=360), LastEngagement.PAST_YEAR, Operator.EXCLUDE, 0, ), ( timezone.now() - timedelta(days=24), LastEngagement.PAST_28_DAYS, Operator.INCLUDE, 1, ), ( timezone.now() - timedelta(days=360), LastEngagement.PAST_6_MONTHS, Operator.EXCLUDE, 1, ), ], ) @pytest.mark.db def test_get_size_with_last_engagement( self, timestamp: datetime.datetime, engagement: LastEngagement, operator: Operator, expected: int, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, last_event_date=timestamp, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( last_engagement=LastEngagementFilter( value=engagement, operator=operator ) ), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("fan_segment", "value", "operator", "expected"), [ ( FanSegment.SUPER_FANS, [FanSegment.ENGAGED_FANS, FanSegment.SUPER_FANS], Operator.INCLUDE, 1, ), ( FanSegment.SUPER_FANS, [FanSegment.ENGAGED_FANS, FanSegment.CASUAL_FANS], Operator.INCLUDE, 0, ), ( FanSegment.CASUAL_FANS, [FanSegment.ENGAGED_FANS, FanSegment.SUPER_FANS], Operator.EXCLUDE, 1, ), ], ) @pytest.mark.db def test_get_size_with_fan_segments( self, fan_segment: FanSegment, value: list[FanSegment], operator: Operator, expected: int, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, segment_name=fan_segment.value, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( fan_segments=FanSegmentInFilter(value=value, operator=operator) ), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("fan_birth_month", "value", "operator", "expected"), [ ( BirthMonth.JAN, [BirthMonth.JAN, BirthMonth.FEB], Operator.INCLUDE, 1, ), ( BirthMonth.JAN, [BirthMonth.FEB, BirthMonth.MAR], Operator.INCLUDE, 0, ), ( BirthMonth.JAN, [BirthMonth.FEB, BirthMonth.MAR], Operator.EXCLUDE, 1, ), ( None, [BirthMonth.JAN, BirthMonth.FEB], Operator.INCLUDE, 0, ), ( None, [BirthMonth.JAN, BirthMonth.FEB], Operator.EXCLUDE, 1, ), ], ) @pytest.mark.db def test_get_size_with_birth_months( self, fan_birth_month: BirthMonth | None, value: list[BirthMonth], operator: Operator, expected: int, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, birth_month=fan_birth_month.value if fan_birth_month else None, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( birth_months=BirthMonthInFilter(value=value, operator=operator) ), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("include_secondary_fans", "fan_segments", "expected"), [ pytest.param( None, None, ["new_fan", "super_fan", "secondary_fan"], id="ALL with undefined include_secondary_fans and fan_segments", ), pytest.param( None, FanSegmentInFilter( value=[FanSegment.SUPER_FANS], operator=Operator.INCLUDE ), ["super_fan"], id="IN (SUPER_FANS)", ), pytest.param( None, FanSegmentInFilter( value=[FanSegment.SUPER_FANS], operator=Operator.EXCLUDE ), ["new_fan", "secondary_fan"], id="NOT IN (SUPER_FANS)", ), pytest.param( False, None, ["new_fan", "super_fan"], id="NOT IN (SECONDARY_FANS)", ), pytest.param( False, FanSegmentInFilter( value=[FanSegment.SUPER_FANS], operator=Operator.INCLUDE ), ["super_fan"], id="IN (SUPER_FANS) AND NOT IN (SECONDARY_FANS)", ), pytest.param( False, FanSegmentInFilter( value=[FanSegment.SUPER_FANS], operator=Operator.EXCLUDE ), ["new_fan"], id="NOT IN (SUPER_FANS) AND NOT IN (SECONDARY_FANS)", ), pytest.param( True, None, ["new_fan", "super_fan", "secondary_fan"], id="ALL with include_secondary_fans=True", ), pytest.param( True, FanSegmentInFilter( value=[FanSegment.SUPER_FANS], operator=Operator.INCLUDE ), ["super_fan", "secondary_fan"], id="IN (SUPER_FANS) OR IN (SECONDARY_FANS)", ), pytest.param( True, FanSegmentInFilter( value=[FanSegment.SUPER_FANS], operator=Operator.EXCLUDE ), ["new_fan", "secondary_fan"], id="NOT IN (SUPER_FANS) AND OTHERS", ), pytest.param( True, FanSegmentInFilter( value=[FanSegment.SECONDARY_FANS], operator=Operator.EXCLUDE ), ["new_fan", "super_fan", "secondary_fan"], id="ALL with include_secondary_fans=True AND NOT IN (SECONDARY_FANS)", ), pytest.param( True, FanSegmentInFilter( value=[FanSegment.SECONDARY_FANS], operator=Operator.INCLUDE ), ["new_fan", "super_fan", "secondary_fan"], id="ALL with include_secondary_fans=True AND IN (SECONDARY_FANS)", ), ], ) @pytest.mark.db def test_get_size_with_fan_segments_and_include_secondary_fans_value( self, include_secondary_fans: bool, fan_segments: FanSegmentInFilter | None, expected: list[str], repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, faker: Faker, identity: Identity, enable_features: EnableFeatures, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, segment_name=FanSegment.NEW_FANS, fan_id="new_fan", ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, segment_name=FanSegment.SUPER_FANS, fan_id="super_fan", ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, segment_name=FanSegment.SECONDARY_FANS, fan_id="secondary_fan", ) with enable_features([AUDIENCE_SHOW_ADD_SECONDARY_FANS_TOGGLE]): result = repository.get_ids( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( fan_segments=fan_segments, include_secondary_fans=include_secondary_fans, ), target=AudienceTarget.ADS, ) ) assert result == IsList(*expected, check_order=False) @pytest.mark.db def test_get_size_with_songwhip_presave_page_filter_include( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_presave_page_fan_dbt = create_reporting_model( SongwhipPresavePageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( songwhipPresavePageIds=SongwhipPresavePageInFilter( value=[songwhip_presave_page_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db def test_get_size_with_songwhip_presave_page_filter_exclude( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_presave_page_fan_dbt = create_reporting_model( SongwhipPresavePageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( songwhipPresavePageIds=SongwhipPresavePageInFilter( value=[songwhip_presave_page_fan_dbt.id], operator=Operator.EXCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 4 @pytest.mark.db def test_get_size_with_songwhip_presave_page_filter_for_vendor( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = 1000 subaccount_id = 2000 global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_presave_page_fan_dbt = create_reporting_model( SongwhipPresavePageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria.model_construct( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), global_participant_id=global_participant_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=0, filters=AudienceFilters.model_construct( songwhipPresavePageIds=SongwhipPresavePageInFilter( value=[songwhip_presave_page_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db def test_get_size_with_songwhip_campaign_ids_filter_include_presave( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_presave_page_fan_dbt = create_reporting_model( SongwhipPresavePageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( songwhipCampaignIds=SongwhipCampaignInFilter( value=[songwhip_presave_page_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db def test_get_size_with_songwhip_campaign_ids_filter_include_custom_page( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_custom_page_fan_dbt = create_reporting_model( SongwhipCustomPageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( songwhipCampaignIds=SongwhipCampaignInFilter( value=[songwhip_custom_page_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db def test_get_size_with_songwhip_campaign_ids_filter_exclude( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_presave_page_fan_dbt = create_reporting_model( SongwhipPresavePageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( songwhipCampaignIds=SongwhipCampaignInFilter( value=[songwhip_presave_page_fan_dbt.id], operator=Operator.EXCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 4 @pytest.mark.db def test_get_size_with_songwhip_campaign_ids_filter_for_vendor( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = 1000 subaccount_id = 2000 global_participant_id = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) songwhip_presave_page_fan_dbt = create_reporting_model( SongwhipPresavePageFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria.model_construct( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), global_participant_id=global_participant_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=0, filters=AudienceFilters.model_construct( songwhipCampaignIds=SongwhipCampaignInFilter( value=[songwhip_presave_page_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db @pytest.mark.parametrize("action", list(EmailCampaignAction)) def test_get_size_with_email_campaigns_filter_include( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, action: EmailCampaignAction, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() campaign_id_1 = fake.uuid4_string() campaign_id_2 = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_1, fan_id=audience_filters_dbt[0].fan_id, delivered=True, opened=True, clicked=True, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_2, fan_id=audience_filters_dbt[1].fan_id, delivered=True, opened=True, clicked=True, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( emailCampaigns=EmailCampaignsFilter( value=[campaign_id_1, campaign_id_2], operator=Operator.INCLUDE, action=action, ) ), target=AudienceTarget.ADS, ) ) assert size == 2 @pytest.mark.db def test_get_size_with_email_campaigns_filter_exclude_delivered( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() campaign_id_1 = fake.uuid4_string() campaign_id_2 = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_1, fan_id=audience_filters_dbt[0].fan_id, delivered=True, opened=True, clicked=True, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_2, fan_id=audience_filters_dbt[1].fan_id, delivered=True, opened=True, clicked=True, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_2, fan_id=audience_filters_dbt[2].fan_id, delivered=False, opened=False, clicked=False, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( emailCampaigns=EmailCampaignsFilter( value=[campaign_id_1, campaign_id_2], operator=Operator.EXCLUDE, action=EmailCampaignAction.DELIVERED, ) ), target=AudienceTarget.ADS, ) ) assert size == 3 @pytest.mark.db @pytest.mark.parametrize( "action", [EmailCampaignAction.OPENED, EmailCampaignAction.CLICKED] ) def test_get_size_with_email_campaigns_filter_not_opened_and_not_clicked( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, action: EmailCampaignAction, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() campaign_id_1 = fake.uuid4_string() campaign_id_2 = fake.uuid4_string() audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_1, fan_id=audience_filters_dbt[0].fan_id, delivered=True, opened=False, clicked=False, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_2, fan_id=audience_filters_dbt[1].fan_id, delivered=False, opened=False, clicked=False, ) create_reporting_model( EmailAnalyticsByCampaignFanDbt, campaign_id=campaign_id_2, fan_id=audience_filters_dbt[2].fan_id, delivered=True, opened=False, clicked=False, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( emailCampaigns=EmailCampaignsFilter( value=[campaign_id_1, campaign_id_2], operator=Operator.EXCLUDE, action=action, ) ), target=AudienceTarget.ADS, ) ) assert size == 2 @pytest.mark.db def test_get_size_with_global_participant_id( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id_1 = fake.uuid4_string() global_participant_id_2 = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id_1, ) # Should not be included in the result create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id_2, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id_1), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db def test_get_size_with_custom_list_id( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() custom_list_id_1 = fake.uuid4_string() custom_list_id_2 = fake.uuid4_string() create_reporting_model( AudienceFiltersCustomListAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, custom_list_id=custom_list_id_1, ) # Should not be included in the result create_reporting_model( AudienceFiltersCustomListAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, custom_list_id=custom_list_id_2, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_custom_list_id(custom_list_id_1), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.parametrize( ("fan_birth_month", "value", "operator", "expected"), [ ( BirthMonth.JAN, [BirthMonth.JAN, BirthMonth.FEB], Operator.INCLUDE, 1, ), ( BirthMonth.JAN, [BirthMonth.FEB, BirthMonth.MAR], Operator.INCLUDE, 0, ), ( BirthMonth.JAN, [BirthMonth.FEB, BirthMonth.MAR], Operator.EXCLUDE, 1, ), ( None, [BirthMonth.JAN, BirthMonth.FEB], Operator.INCLUDE, 0, ), ( None, [BirthMonth.JAN, BirthMonth.FEB], Operator.EXCLUDE, 1, ), ], ) @pytest.mark.db def test_get_size_with_custom_list_and_birth_months( self, fan_birth_month: BirthMonth | None, value: list[BirthMonth], operator: Operator, expected: int, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() custom_list_id = fake.uuid4_string() create_reporting_model( AudienceFiltersCustomListAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, custom_list_id=custom_list_id, birth_month=fan_birth_month.value if fan_birth_month else None, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_custom_list_id(custom_list_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( birth_months=BirthMonthInFilter(value=value, operator=operator) ), target=AudienceTarget.ADS, ) ) assert size == expected @pytest.mark.parametrize( ("fan_region_code", "fan_region", "fan_country_iso2", "regions", "expected"), [ # US: exact match on region_code ( "CA", "California", "US", [ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], 1, ), ( "TX", "Texas", "US", [ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], 0, ), ( None, None, "US", [ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], 0, ), # non-US: similarity match on region name ( None, "Catalonia", "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 1, ), ( None, "Cataloniia", "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 1, ), ( None, "Andalusia", "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 0, ), ( None, "Catalonia", "GB", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 0, ), ( None, None, "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 0, ), ], ) @pytest.mark.db def test_get_size_with_regions( self, fan_region_code: str | None, fan_region: str | None, fan_country_iso2: str, regions: list[RegionFilter], expected: int, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, region_code=fan_region_code, region=fan_region, country_iso2=fan_country_iso2, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(regions=regions), target=AudienceTarget.ADS, ) ) assert result == expected @pytest.mark.parametrize( ("fan_region_code", "fan_region", "fan_country_iso2", "regions", "expected"), [ # US: exact match on region_code ( "CA", "California", "US", [ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], 1, ), ( "TX", "Texas", "US", [ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], 0, ), ( None, None, "US", [ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], 0, ), # non-US: similarity match on region name ( None, "Catalonia", "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 1, ), ( None, "Cataloniia", "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 1, ), ( None, "Andalusia", "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 0, ), ( None, "Catalonia", "GB", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 0, ), ( None, None, "ES", [RegionFilter(placeId="2", title="Catalonia", countryCode="ES")], 0, ), ], ) @pytest.mark.db def test_get_size_with_custom_list_and_regions( self, fan_region_code: str | None, fan_region: str | None, fan_country_iso2: str, regions: list[RegionFilter], expected: int, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() custom_list_id = fake.uuid4_string() create_reporting_model( AudienceFiltersCustomListAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, custom_list_id=custom_list_id, region_code=fan_region_code, region=fan_region, country_iso2=fan_country_iso2, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_custom_list_id(custom_list_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(regions=regions), target=AudienceTarget.ADS, ) ) assert size == expected @pytest.mark.db def test_get_size_with_regions_us_matches_only_exact_region_code_from_multiple_fans( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() for region_code, region, country in [ ("CA", "California", "US"), ("TX", "Texas", "US"), ("NY", "New York", "US"), ]: create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, region_code=region_code, region=region, country_iso2=country, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct( regions=[ RegionFilter( placeId="1", title="California", countryCode="US", regionCode="CA", ) ], ), target=AudienceTarget.ADS, ) ) assert result == 1 @pytest.mark.db def test_get_size_with_regions_non_us_similarity_matches_but_dissimilar_does_not( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() # "Cataloniia" is a typo close enough to match; "Andalusia" is too different for region, country in [("Cataloniia", "ES"), ("Andalusia", "ES")]: create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, region=region, country_iso2=country, ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct( regions=[ RegionFilter(placeId="2", title="Catalonia", countryCode="ES") ], ), target=AudienceTarget.ADS, ) ) assert result == 1 @pytest.mark.db def test_get_size_with_global_participant_id_and_custom_list_id( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id_1 = fake.uuid4_string() global_participant_id_2 = fake.uuid4_string() custom_list_id_1 = fake.uuid4_string() custom_list_id_2 = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id_1, ) create_reporting_model( AudienceFiltersCustomListAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, custom_list_id=custom_list_id_1, ) # Should not be included in the result create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id_2, ) create_reporting_model( AudienceFiltersCustomListAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, custom_list_id=custom_list_id_2, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id_1), FanDataListId.from_custom_list_id(custom_list_id_1), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ) ) assert size == 2 @pytest.mark.db def test_get_size_is_global_empty_with_account_specific( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() # Should not be included in the result create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, is_global=True, ) ) assert size == 0 @pytest.mark.db def test_get_size_is_global_empty_no_reps( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, is_global=True, ) ) assert size == 0 @pytest.mark.db def test_get_size_is_global_with_main_rep_only( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( ArtistRosterMainRep, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( GlobalAudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, is_global=True, ) ) assert size == 1 @pytest.mark.db def test_get_size_is_global_with_local_rep_only( self, repository: AudienceFanRepository, identity_id: str, faker: Faker, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_code = faker.country_code() create_reporting_model( ArtistRosterLocalRep, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_code=country_code, ) create_reporting_model( GlobalAudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, country_iso2=country_code, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, is_global=True, ) ) assert size == 1 @pytest.mark.db def test_get_size_is_global_with_local_rep_only_and_countries( self, repository: AudienceFanRepository, identity_id: str, faker: Faker, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_code_1 = faker.country_code() country_code_2 = faker.country_code() country_code_3 = faker.country_code() create_reporting_model( ArtistRosterLocalRep, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_code=country_code_1, ) create_reporting_model( ArtistRosterLocalRep, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_code=country_code_2, ) create_reporting_model( GlobalAudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, country_iso2=country_code_1, ) create_reporting_model( GlobalAudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, country_iso2=country_code_2, ) # Should not be included in the result create_reporting_model( GlobalAudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, country_iso2=country_code_3, ) size = repository.get_size( AudienceCriteria( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( countries=[country_code_1, country_code_2, country_code_3] ), target=AudienceTarget.ADS, is_global=True, ) ) assert size == 2 # -- Email target size -- @pytest.mark.db def test_get_size_with_target_email( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) size = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.EMAIL, ) ) assert size == 2 # -- ADS target insert -- @pytest.mark.db def test_insert_bulk( self, repository: AudienceFanRepository, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: size = 3 global_participant_id = fake.uuid4_string() vendor_id = fake.integer() subaccount_id = fake.integer() snapshot_id = fake.uuid4_string() create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, is_primary_fan=False, ) repository.insert_bulk( snapshot_id, criteria=AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ), ) assert repository.count() == size @pytest.mark.db def test_insert_bulk_internal_employee( self, repository: AudienceFanRepository, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, fake: FakerTyped, internal_employee_identity: Identity, ) -> None: size = 3 global_participant_id = fake.uuid4_string() vendor_id = fake.integer() subaccount_id = fake.integer() snapshot_id = fake.uuid4_string() create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( AudienceFiltersAdCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, is_primary_fan=False, ) repository.insert_bulk( snapshot_id, criteria=AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=internal_employee_identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.ADS, ), ) assert repository.count() == 4 # -- Email target insert -- @pytest.mark.db def test_insert_bulk_email_target( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, identity: Identity, ) -> None: global_participant_id = fake.uuid4_string() vendor_id = fake.integer() subaccount_id = fake.integer() snapshot_id = fake.uuid4_string() custom_list_id = fake.pystr() fan_without_engagement = create_reporting_model( AudienceFiltersEmailCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, last_email_event_date=None, ) fan = create_reporting_model( AudienceFiltersEmailCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, last_email_event_date=datetime.datetime(2022, 2, 24, tzinfo=datetime.UTC), ) create_reporting_model( AudienceFiltersCustomListEmailCampaignDbt, fan_id=fan.fan_id, custom_list_id=custom_list_id, vendor_id=vendor_id, subaccount_id=subaccount_id, last_email_event_date=datetime.datetime(2020, 1, 1, tzinfo=datetime.UTC), ) repository.insert_bulk( snapshot_id, criteria=AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), FanDataListId.from_custom_list_id(custom_list_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.EMAIL, ), ) assert repository.all() == IsList( AudienceFan( snapshot_id=snapshot_id, fan_id=fan.fan_id, last_engagement_at=fan.last_email_event_date, ), AudienceFan(snapshot_id=snapshot_id, fan_id=fan_without_engagement.fan_id), check_order=False, ) # -- ADS target export -- @pytest.mark.parametrize( "reason", ( AudienceExportReason.ADS_META, AudienceExportReason.ADS_TIKTOK, AudienceExportReason.ADS_GOOGLE, AudienceExportReason.ADS_SNAPCHAT, AudienceExportReason.ADS_SPOTIFY, ), ) @pytest.mark.db def test_get_export_fan_data_ads_target_email( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, reason: AudienceExportReason, identity: Identity, fake: FakerTyped, ) -> None: email = fake.email() audience_fan = create_reporting_model(AudienceFan) create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan.fan_id, fan_email=email, fan_mobile_phone=None, ) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=reason, identity=identity, ) assert rows == [ { "email": email, "phone": IsStr(), } ] @pytest.mark.parametrize( "reason", ( AudienceExportReason.ADS_META, AudienceExportReason.ADS_SNAPCHAT, ), ) @pytest.mark.db def test_get_export_fan_data_ads_target_phone_without_plus( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, reason: AudienceExportReason, identity: Identity, ) -> None: phone = "+1234567890" audience_fan = create_reporting_model(AudienceFan) create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan.fan_id, fan_mobile_phone=phone ) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=reason, identity=identity, ) assert rows == [ { "email": IsStr(), "phone": phone.replace("+", ""), } ] @pytest.mark.parametrize( "reason", ( AudienceExportReason.ADS_TIKTOK, AudienceExportReason.ADS_GOOGLE, AudienceExportReason.ADS_SPOTIFY, ), ) @pytest.mark.db def test_get_export_fan_data_ads_target_phone_with_plus( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, reason: AudienceExportReason, identity: Identity, ) -> None: phone = "+1234567890" audience_fan = create_reporting_model(AudienceFan) create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan.fan_id, fan_mobile_phone=phone ) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=reason, identity=identity, ) assert rows == [ { "email": IsStr(), "phone": phone, } ] # -- Text target export -- @pytest.mark.parametrize( "reason", ( AudienceExportReason.TEXT_COMMUNITY, AudienceExportReason.TEXT_LAYOUT, AudienceExportReason.TEXT_SUBTEXT, ), ) @pytest.mark.db def test_get_export_fan_data_text_target( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, reason: AudienceExportReason, identity: Identity, ) -> None: audience_fan = create_reporting_model(AudienceFan) create_reporting_model(FanPersonalDataDbt, fan_id=audience_fan.fan_id) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=reason, identity=identity, ) assert rows == [ { "fan_id": audience_fan.fan_id, } ] # -- Email target export -- @pytest.mark.db def test_get_export_fan_data_email_target_salesforce( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, ) -> None: audience_fan = create_reporting_model(AudienceFan) personal_data = create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan.fan_id, ) fan_location = create_reporting_model( FanLocationDbt, fan_id=audience_fan.fan_id ) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=AudienceExportReason.EMAIL_SALESFORCE, identity=identity, ) assert rows == [ { "first_name": personal_data.fan_first_name, "last_name": personal_data.fan_last_name, "country": fan_location.country_iso2, "email": personal_data.fan_email, } ] @pytest.mark.db def test_get_export_fan_data_email_target_exclude_phone_only_fans( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, ) -> None: audience_fan1 = create_reporting_model(AudienceFan) audience_fan2 = create_reporting_model( AudienceFan, snapshot_id=audience_fan1.snapshot_id ) audience_fan3 = create_reporting_model( AudienceFan, snapshot_id=audience_fan1.snapshot_id ) audience_fan4 = create_reporting_model( AudienceFan, snapshot_id=audience_fan1.snapshot_id ) audience_fan5 = create_reporting_model( AudienceFan, snapshot_id=audience_fan1.snapshot_id ) create_reporting_model(FanPersonalDataDbt, fan_id=audience_fan1.fan_id) create_reporting_model(FanPersonalDataDbt, fan_id=audience_fan2.fan_id) create_reporting_model(FanPersonalDataDbt, fan_id=audience_fan3.fan_id) create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan4.fan_id, fan_email=None ) create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan5.fan_id, fan_email=None ) rows = repository.get_export_fan_data( snapshot_id=audience_fan1.snapshot_id, reason=AudienceExportReason.EMAIL_SALESFORCE, identity=identity, ) assert len(rows) == 3 # -- Common export helpers -- @pytest.mark.db def test_get_export_fan_data_no_personal_data( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, ) -> None: audience_fan = create_reporting_model(AudienceFan) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=AudienceExportReason.ADS_META, identity=identity, ) assert rows == [] # -- ADS target share -- @pytest.mark.db def test_get_share_fan_data_for_meta( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, faker: Faker, ) -> None: audience_fan = create_reporting_model(AudienceFan) email = faker.email() phone = faker.phone_number() create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan.fan_id, fan_email=email, fan_mobile_phone=phone, ) rows = repository.get_share_fan_data( snapshot_id=audience_fan.snapshot_id, platform=AudienceSharePlatform.META, ) assert rows == [ { "fan_id": make_fan_id(email), "fan_phone": make_fan_id(phone), } ] @pytest.mark.db def test_get_share_fan_data_for_google( self, audience_fan_repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, fake: FakerTyped, ) -> None: audience_fan = create_reporting_model(AudienceFan) email = fake.uuid4_string() phone = fake.uuid4_string() create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan.fan_id, fan_email=email, fan_mobile_phone=phone, ) rows = audience_fan_repository.get_share_fan_data( snapshot_id=audience_fan.snapshot_id, platform=AudienceSharePlatform.GOOGLE, ) assert rows == [ { "fan_id": make_fan_id(email), "fan_phone": make_fan_id(phone), } ] @pytest.mark.db def test_get_share_fan_data_no_personal_data( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, ) -> None: audience_fan = create_reporting_model(AudienceFan) rows = repository.get_export_fan_data( snapshot_id=audience_fan.snapshot_id, reason=AudienceExportReason.ADS_META, identity=identity, ) assert rows == [] @pytest.mark.db def test_get_share_fan_data_phone_only_ok( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, faker: Faker, ) -> None: audience_fan1 = create_reporting_model(AudienceFan) audience_fan2 = create_reporting_model( AudienceFan, snapshot_id=audience_fan1.snapshot_id ) email = faker.email() fan1_phone = faker.phone_number() fan2_phone = faker.phone_number() create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan1.fan_id, fan_email=email, fan_mobile_phone=fan1_phone, ) create_reporting_model( FanPersonalDataDbt, fan_id=audience_fan2.fan_id, fan_email=None, fan_mobile_phone=fan2_phone, ) rows = repository.get_share_fan_data( snapshot_id=audience_fan1.snapshot_id, platform=AudienceSharePlatform.META, ) assert rows == IsList( { "fan_id": make_fan_id(email), "fan_phone": make_fan_id(fan1_phone.lstrip("+0")), }, { "fan_id": None, "fan_phone": make_fan_id(fan2_phone.lstrip("+0")), }, check_order=False, ) # -- ADS target CRM filters -- @pytest.mark.db def test_get_size_with_crm_campaign_id_filter_include( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() custom_list_id = None audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) crm_campaign_id_fan_dbt = create_reporting_model( CrmCampaignIdFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, custom_list_id=custom_list_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( crmCampaigns=CrmCampaignFilter( value=[crm_campaign_id_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 @pytest.mark.db def test_get_size_with_crm_campaign_id_filter_exclude( self, repository: AudienceFanRepository, identity: Identity, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = fake.integer() subaccount_id = fake.integer() global_participant_id = fake.uuid4_string() custom_list_id = None audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) crm_campaign_id_fan_dbt = create_reporting_model( CrmCampaignIdFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, custom_list_id=custom_list_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria( identity=identity, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=subaccount_id, filters=AudienceFilters.model_construct( crmCampaigns=CrmCampaignFilter( value=[crm_campaign_id_fan_dbt.id], operator=Operator.EXCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 4 @pytest.mark.db def test_get_size_with_crm_campaign_id_filter_for_vendor( self, repository: AudienceFanRepository, identity_id: str, fake: FakerTyped, create_reporting_model_batch: CreateReportingModelBatch, create_reporting_model: CreateReportingModel, ) -> None: size = 5 vendor_id = 1000 subaccount_id = 2000 global_participant_id = fake.uuid4_string() custom_list_id = None audience_filters_dbt = create_reporting_model_batch( AudienceFiltersAdCampaignDbt, size=size, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) crm_campaign_id_fan_dbt = create_reporting_model( CrmCampaignIdFanDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, custom_list_id=custom_list_id, fan_id=audience_filters_dbt[0].fan_id, ) size = repository.get_size( AudienceCriteria.model_construct( identity=Identity( id=identity_id, brand="sme", is_internal_employee=True ), global_participant_id=global_participant_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], vendor_id=vendor_id, subaccount_id=0, filters=AudienceFilters.model_construct( crmCampaigns=CrmCampaignFilter( value=[crm_campaign_id_fan_dbt.id], operator=Operator.INCLUDE, ) ), target=AudienceTarget.ADS, ) ) assert size == 1 class TestAudienceFanRepositoryTextTarget: @pytest.mark.db def test_get_size_with_target_text( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, faker: Faker, identity: Identity, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=artist_phone_number, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_out", ) other_global_participant_id = faker.uuid4() other_phone_number = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=other_global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=other_global_participant_id, phone_number=other_phone_number, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=other_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 2 @pytest.mark.db def test_get_size_text_with_country_filters( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterLocalRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=artist_phone_number, ) included_country = "US" excluded_country = "GB" create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), fan_country=included_country, subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), fan_country=excluded_country, subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), fan_country="CA", subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( countries=[included_country, excluded_country] ), target=AudienceTarget.TEXT, excluded_countries=[excluded_country], ) ) assert result == 1 @pytest.mark.db def test_get_size_text_with_other_filters_returns_zero( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=artist_phone_number, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id), ], identity=identity, filters=AudienceFilters.model_construct( dsp=DigitalServiceProviderInFilter( value=[DigitalServiceProvider.SPOTIFY], operator=Operator.INCLUDE, ), ), target=AudienceTarget.TEXT, ) ) assert result == 0 @pytest.mark.db def test_get_size_text_only_excluded_countries( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=artist_phone_number, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), fan_country="US", subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), fan_country="GB", subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, excluded_countries=["GB"], ) ) assert result == 1 @pytest.mark.db def test_get_size_text_fan_without_country_is_included( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=artist_phone_number, ) # No country create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_multiple_artists_does_not_cross_mix( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_1 = faker.uuid4() global_participant_2 = faker.uuid4() artist_phone_number_1 = faker.phone_number() artist_phone_number_2 = faker.phone_number() # Artist A create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_1, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_1, phone_number=artist_phone_number_1, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number_1, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) # Artist B create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_2, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_2, phone_number=artist_phone_number_2, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number_2, fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_1)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_deduplicates_multiple_twilio_rows( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant, phone_number=artist_phone_number, ) fan_phone = faker.phone_number() create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=fan_phone, subscription_status="opted_in", channel="SMS", ) # duplicate create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=fan_phone, subscription_status="opted_in", channel="WHATSAPP", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_local_rep_only( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() artist_phone_number = faker.phone_number() create_reporting_model( ArtistRosterLocalRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, country_code="US", ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=artist_phone_number, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number, fan_phone_number=faker.phone_number(), subscription_status="opted_in", fan_country="US", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_excludes_unsubscribed( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() phone = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=phone, ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=phone, fan_phone_number=faker.phone_number(), subscription_status="opted_out", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 0 @pytest.mark.db def test_get_size_text_twilio_and_filters_dont_overlap( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() phone = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=phone, ) # Twilio fan twilio_phone = faker.phone_number() create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=phone, fan_phone_number=twilio_phone, subscription_status="opted_in", ) # all_filters contains other fan_id → must NOT add anything create_reporting_model( AudienceFiltersTextCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, fan_id="otherfan", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_filter_but_no_twilio_subscription( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) # In filters, but NO Twilio subscription → ignore create_reporting_model( AudienceFiltersTextCampaignDbt, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, fan_id="fan123", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 0 @pytest.mark.db def test_get_size_text_multiple_artist_numbers( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() phone1 = faker.phone_number() phone2 = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=phone1, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=phone2, ) fan_phone = faker.phone_number() create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=phone1, fan_phone_number=fan_phone, subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=phone2, fan_phone_number=fan_phone, subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_trims_phone_match( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() phone = " +12345 " create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id, phone_number=phone, # stored with spaces ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number="+12345", # trimmed fan_phone_number=faker.phone_number(), subscription_status="opted_in", ) result = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[FanDataListId.from_artist_id(global_participant_id)], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert result == 1 @pytest.mark.db def test_get_size_text_fan_connected_to_two_artists_but_only_one_selected( self, repository: AudienceFanRepository, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id_1 = faker.uuid4() global_participant_id_2 = faker.uuid4() artist_phone_number_1 = faker.phone_number() artist_phone_number_2 = faker.phone_number() fan_phone = faker.phone_number() create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id_1, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id_1, phone_number=artist_phone_number_1, ) create_reporting_model( ArtistRosterMainRep, global_participant_id=global_participant_id_2, vendor_id=vendor_id, subaccount_id=subaccount_id, ) create_reporting_model( ArtistPhoneNumber, global_participant_id=global_participant_id_2, phone_number=artist_phone_number_2, ) # Fan is subscribed to both artists create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number_1, fan_phone_number=fan_phone, subscription_status="opted_in", ) create_reporting_model( TwilioFanSubscriptionStatus, artist_phone_number=artist_phone_number_2, fan_phone_number=fan_phone, subscription_status="opted_in", ) # Query only artist 1 size = repository.get_size( AudienceCriteria( vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_list_ids=[ FanDataListId.from_artist_id(global_participant_id_1) ], identity=identity, filters=AudienceFilters.model_construct(), target=AudienceTarget.TEXT, ) ) assert size == 1