from unittest import mock import pytest from dirty_equals import IsList from faker import Faker from fansifter_common.auth.identity import Identity from fansifter_common.utils import timezone from dmp.adapters.aws.location import Location from dmp.audiences.dtos import AudienceCriteria from dmp.audiences.enums import AudienceTarget from dmp.audiences.exceptions import AudienceNotFoundError from dmp.audiences.filters import AudienceFilters, CityFilter from dmp.audiences.models import Audience, AudienceFanDataList from dmp.audiences.services import AudienceService from dmp.audiences.types import FanDataListId from dmp.fandata.models import ( AudienceFiltersAdCampaignDbt, AudienceFiltersEmailCampaignDbt, EmailDomainToISPMapping, FanPersonalDataDbt, GlobalAudienceFiltersEmailCampaignDbt, ) from dmp.rosters.enums import FanDataListType from dmp.rosters.exceptions import MainRepArtistOnlyAllowedError from dmp.rosters.models import ArtistRosterLocalRep, ArtistRosterMainRep from tests.unit.types import CreateModel, CreateReportingModel, OverrideSettings class TestAudienceService: @pytest.mark.db def test_get_audience( self, service: AudienceService, create_model: CreateModel ) -> None: audience = create_model(Audience) result = service.get_audience(audience.id) assert result == audience @pytest.mark.db def test_get_audience_archived( self, service: AudienceService, create_model: CreateModel, faker: Faker, ) -> None: audience = create_model( Audience, archived_at=timezone.now(), archived_by=faker.uuid4(), ) with pytest.raises(AudienceNotFoundError): service.get_audience(audience.id) @pytest.mark.db def test_refresh_audience_snapshot( self, service: AudienceService, create_model: CreateModel, identity: Identity, ) -> None: audience = create_model(Audience, target=AudienceTarget.ADS) service.refresh_audience(audience, identity=identity, is_global=False) snapshot = audience.latest_snapshot assert snapshot assert snapshot.target == audience.target assert snapshot.filters == audience.filters assert snapshot.created_by == identity.id assert audience.refreshed_by == identity.id assert audience.recipients_count_by_email_provider is None @pytest.mark.db def test_refresh_audience_snapshot_with_global_excluded_country( self, service: AudienceService, create_model: CreateModel, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: country_es = "ES" country_us = "US" global_participant_id = faker.uuid4() audience = create_model( Audience, filters=AudienceFilters.model_construct(), fandata_lists=[ AudienceFanDataList( list_id=global_participant_id, type=FanDataListType.ARTIST, ) ], target=AudienceTarget.ADS, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=global_participant_id, country_iso2=country_es, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=audience.global_participant_id, country_iso2=country_us, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, global_participant_id=audience.global_participant_id, country_iso2=country_us, ) with override_settings(audience_global_excluded_countries=[country_es]): service.refresh_audience(audience, identity=identity, is_global=False) snapshot = audience.latest_snapshot assert snapshot assert audience.fan_count == snapshot.fan_count == 2 @pytest.mark.db def test_get_audience_size_with_global_excluded_countries( self, service: AudienceService, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_es = "ES" country_us = "US" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_es, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_us, ) with override_settings(audience_global_excluded_countries=[country_es]): size = service.get_audience_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(), target=AudienceTarget.ADS, ), ) assert size == 1 @pytest.mark.db def test_get_audience_size_with_global_excluded_countries_and_ads_countries( self, service: AudienceService, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_es = "ES" country_us = "US" country_hu = "HU" country_pt = "PT" create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_es, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_us, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_hu, ) create_reporting_model( AudienceFiltersAdCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_pt, ) with override_settings( audience_global_excluded_countries=[country_es], audience_ads_excluded_countries=[country_hu], ): size = service.get_audience_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( countries=["ES", "HU", "US", "PT"] ), target=AudienceTarget.ADS, ), ) assert size == 2 @pytest.mark.db def test_get_audience_size_with_global_excluded_countries_text_target( self, service: AudienceService, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_es = "ES" country_us = "US" country_de = "DE" create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_es, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_us, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_de, ) with override_settings( audience_global_excluded_countries=[country_es], audience_ads_excluded_countries=[country_de], ): size = service.get_audience_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=["ES", "DE", "US"] ), target=AudienceTarget.EMAIL, ), ) assert size == 2 @pytest.mark.db def test_get_audience_size_exclude_france_when_not_selected( self, service: AudienceService, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_fr = "FR" country_us = "US" country_de = "DE" create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_fr, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_us, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_de, ) with override_settings( audience_global_excluded_countries=[], ): size = service.get_audience_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 @pytest.mark.db def test_get_audience_size_include_france_when_selected( self, service: AudienceService, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_fr = "FR" country_us = "US" create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_fr, ) create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_us, ) with override_settings( audience_global_excluded_countries=[], ): size = service.get_audience_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_fr, country_us, ] ), target=AudienceTarget.EMAIL, ), ) assert size == 2 @pytest.mark.db def test_get_audience_size_include_france_when_selected_in_cities( self, service: AudienceService, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() country_fr = "FR" city_paris = "1" create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, country_iso2=country_fr, city_id=city_paris, ) with override_settings( audience_global_excluded_countries=[], ): size = service.get_audience_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_paris, country_iso2=country_fr, ) ] ), target=AudienceTarget.EMAIL, ), ) assert size == 1 @pytest.mark.db def test_clean_filters( self, service: AudienceService, location_client_mock: mock.MagicMock ) -> None: filters = AudienceFilters.model_construct( cities=[CityFilter.model_construct(place_id="1", title="Paris")] ) location_client_mock.get_location.return_value = Location.model_validate( { "PlaceId": "1", "Title": "Paris", "Address": { "Country": {"Code2": "FR", "Name": "France"}, "Locality": "Paris", }, "Position": [-74, 40], "MapView": [-74.1, 40.1, -73.9, 39.9], } ) service.clean_cities_filter(filters) assert filters.cities assert filters.cities[0].country_iso2 == "FR" assert filters.cities[0].city_id == "-1" @pytest.mark.db def test_create_audience_is_global_local_rep_is_not_allowed( self, service: AudienceService, identity: Identity, faker: Faker, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() create_reporting_model( AudienceFiltersEmailCampaignDbt, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) create_reporting_model( ArtistRosterLocalRep, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) with pytest.raises(MainRepArtistOnlyAllowedError) as exc_info: _ = service.get_audience_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(), target=AudienceTarget.EMAIL, is_global=True, ), ) assert exc_info.value.global_participant_id == global_participant_id @pytest.mark.db def test_create_audience_is_global_main_rep_allowed( self, service: AudienceService, identity: Identity, faker: Faker, create_reporting_model: CreateReportingModel, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() create_reporting_model( GlobalAudienceFiltersEmailCampaignDbt, global_participant_id=global_participant_id, ) create_reporting_model( ArtistRosterMainRep, vendor_id=vendor_id, subaccount_id=subaccount_id, global_participant_id=global_participant_id, ) size = service.get_audience_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(), target=AudienceTarget.EMAIL, is_global=True, ), ) assert size == 1 @pytest.mark.db def test_refresh_audience_email_calculates_providers( self, service: AudienceService, create_model: CreateModel, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: global_participant_id = faker.uuid4() audience = create_model( Audience, target=AudienceTarget.EMAIL, fandata_lists=[ AudienceFanDataList( list_id=global_participant_id, type=FanDataListType.ARTIST, ) ], ) create_reporting_model(EmailDomainToISPMapping, domain="gmail.com", isp="gmail") for i in range(3): fan = create_reporting_model( AudienceFiltersEmailCampaignDbt, global_participant_id=global_participant_id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_reporting_model( FanPersonalDataDbt, fan_id=fan.fan_id, fan_email=f"user{i}@gmail.com", fan_email_domain="gmail.com", ) fan = create_reporting_model( AudienceFiltersEmailCampaignDbt, global_participant_id=global_participant_id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_reporting_model( FanPersonalDataDbt, fan_id=fan.fan_id, fan_email="user@example.com", fan_email_domain="example.com", ) service.refresh_audience(audience, identity=identity, is_global=False) assert audience.recipients_count_by_email_provider == { "gmail": 3, "OTHER": 1, } @pytest.mark.db def test_upsert_audience_fans_email_calculates_providers( self, service: AudienceService, create_model: CreateModel, create_reporting_model: CreateReportingModel, identity: Identity, faker: Faker, ) -> None: global_participant_id = faker.uuid4() audience = create_model( Audience, target=AudienceTarget.EMAIL, fandata_lists=[ AudienceFanDataList( list_id=global_participant_id, type=FanDataListType.ARTIST, ) ], ) create_reporting_model(EmailDomainToISPMapping, domain="yahoo.com", isp="yahoo") for i in range(2): fan = create_reporting_model( AudienceFiltersEmailCampaignDbt, global_participant_id=global_participant_id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_reporting_model( FanPersonalDataDbt, fan_id=fan.fan_id, fan_email=f"user{i}@yahoo.com", fan_email_domain="yahoo.com", ) service.upsert_audience_fans(audience, identity=identity, is_global=False) assert audience.recipients_count_by_email_provider == {"yahoo": 2} def test_clean_criteria_text_intersects_allowed_countries( self, service: AudienceService, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() allowed_countries = ["US", "CA"] requested_countries = ["US", "FR", "CA"] 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(countries=requested_countries), target=AudienceTarget.TEXT, ) with override_settings(audience_text_allowed_countries=allowed_countries): cleaned = service.clean_criteria(criteria) assert cleaned.filters.countries == IsList("US", "CA", check_order=False) def test_clean_criteria_text_drops_disallowed_countries( self, service: AudienceService, identity: Identity, faker: Faker, override_settings: OverrideSettings, ) -> None: vendor_id = faker.pyint() subaccount_id = faker.pyint() global_participant_id = faker.uuid4() allowed_countries = ["US"] requested_countries = ["FR", "DE"] 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(countries=requested_countries), target=AudienceTarget.TEXT, ) with override_settings(audience_text_allowed_countries=allowed_countries): cleaned = service.clean_criteria(criteria) assert cleaned.filters.countries == ["XY"]