from datetime import timedelta import pytest from dirty_equals import IsList from fansifter_common.auth.account import Account from fansifter_common.utils import timezone from dmp.adapters.features import AUDIENCE_SHOW_SMS_CAMPAIGNS_PAGE from dmp.audiences.dtos import AudiencesQueryCriteria from dmp.audiences.enums import AudienceTarget from dmp.audiences.models import Audience, AudienceFanDataList from dmp.audiences.repositories import AudienceRepository from dmp.rosters.enums import FanDataListType from tests.unit.types import CreateModel, EnableFeatures class TestAudienceRepository: @pytest.mark.db def test_exists_by_name_and_account( self, repository: AudienceRepository, create_model: CreateModel ) -> None: audience = create_model(Audience) exists = repository.exists_by_name_and_account( name=audience.name.upper(), account=audience.account ) assert exists @pytest.mark.db def test_exists_by_name_and_account_not_exist( self, repository: AudienceRepository, ) -> None: exists = repository.exists_by_name_and_account( name="name", account=Account(vendor_id=10, subaccount_id=0) ) assert not exists @pytest.mark.db def test_find_by_criteria_empty( self, repository: AudienceRepository, identity_id: str, ) -> None: result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[], ad_target_subaccount_ids=[], email_target_vendor_ids=[], email_target_subaccount_ids=[], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [] @pytest.mark.db def test_find_by_criteria( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: audience = create_model(Audience) create_model(Audience) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[audience.vendor_id], ad_target_subaccount_ids=[audience.subaccount_id], email_target_vendor_ids=[audience.vendor_id], email_target_subaccount_ids=[audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [audience] @pytest.mark.db def test_find_by_criteria_different_access_to_vendors_in_target( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: now = timezone.now() email_audience = create_model( Audience, target=AudienceTarget.EMAIL, created_at=now - timedelta(days=1), ) ad_audience = create_model(Audience, target=AudienceTarget.ADS, created_at=now) create_model(Audience) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[ad_audience.vendor_id], ad_target_subaccount_ids=[ad_audience.subaccount_id], email_target_vendor_ids=[email_audience.vendor_id], email_target_subaccount_ids=[email_audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == IsList(ad_audience, email_audience) @pytest.mark.db def test_find_by_criteria_different_access_to_vendors_in_target_with_text_target_access( self, repository: AudienceRepository, create_model: CreateModel, enable_features: EnableFeatures, identity_id: str, ) -> None: now = timezone.now() email_audience = create_model( Audience, target=AudienceTarget.EMAIL, created_at=now - timedelta(days=1), ) ad_audience = create_model(Audience, target=AudienceTarget.ADS, created_at=now) create_model(Audience) text_audience = create_model( Audience, target=AudienceTarget.TEXT, created_at=now - timedelta(days=2) ) with enable_features([AUDIENCE_SHOW_SMS_CAMPAIGNS_PAGE]): result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[ad_audience.vendor_id], ad_target_subaccount_ids=[ad_audience.subaccount_id], email_target_vendor_ids=[email_audience.vendor_id], email_target_subaccount_ids=[email_audience.subaccount_id], text_target_vendor_ids=[text_audience.vendor_id], text_target_subaccount_ids=[text_audience.subaccount_id], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == IsList(ad_audience, email_audience, text_audience) @pytest.mark.db def test_find_by_criteria_with_archived( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: audience = create_model(Audience) # Create an archived audience create_model( Audience, archived_at=timezone.now(), archived_by=identity_id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[audience.vendor_id], ad_target_subaccount_ids=[audience.subaccount_id], email_target_vendor_ids=[audience.vendor_id], email_target_subaccount_ids=[audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [audience] @pytest.mark.db def test_find_by_criteria_search( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: audience = create_model( Audience, name="Artist ABC Estonia", ) create_model( Audience, name="Artist DEF United States", vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[audience.vendor_id], ad_target_subaccount_ids=[audience.subaccount_id], email_target_vendor_ids=[audience.vendor_id], email_target_subaccount_ids=[audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search="abc", global_participant_id=None, custom_list_id=None, target=None, audience_ids=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [audience] @pytest.mark.db def test_find_by_criteria_by_malformed_name( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: audience = create_model(Audience, name="Artist ABC Estonia") result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[audience.vendor_id], ad_target_subaccount_ids=[audience.subaccount_id], email_target_vendor_ids=[audience.vendor_id], email_target_subaccount_ids=[audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search="abc'", global_participant_id=None, custom_list_id=None, target=None, audience_ids=None, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [] @pytest.mark.db def test_find_by_criteria_audience_ids( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 now = timezone.now() # Here we're adding created_at with predefined values to be ensure # that models will not be created at the same time # and ordering in final list will be correct audience_1 = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, created_at=now + timedelta(seconds=1), ) audience_2 = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, created_at=now, ) _audience_3 = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, target=None, audience_ids=[audience_1.id, audience_2.id], ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == IsList(audience_1, audience_2) @pytest.mark.db def test_find_by_criteria_include_archived( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 audience_1 = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id ) audience_2 = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, archived_at=timezone.now(), ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, include_archived=True, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == IsList(audience_1, audience_2, check_order=False) @pytest.mark.db def test_find_by_criteria_target( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, target=AudienceTarget.ADS, ) audience_2 = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, target=AudienceTarget.EMAIL, ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=AudienceTarget.EMAIL, include_archived=True, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [audience_2] @pytest.mark.db def test_find_by_criteria_global_participant_id( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 global_participant_id = "123" audience = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_lists=[ AudienceFanDataList( list_id=global_participant_id, type=FanDataListType.ARTIST ) ], ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=global_participant_id, custom_list_id=None, audience_ids=None, target=None, include_archived=True, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [audience] @pytest.mark.db def test_find_by_criteria_custom_list_id( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 custom_list_id = "123" audience = create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_lists=[ AudienceFanDataList( list_id=custom_list_id, type=FanDataListType.CUSTOM_LIST ) ], ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=custom_list_id, audience_ids=None, target=None, include_archived=True, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [audience] @pytest.mark.db def test_find_by_criteria_custom_list_id_not_found( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 global_participant_id = "123" create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, fandata_lists=[ AudienceFanDataList( list_id=global_participant_id, type=FanDataListType.ARTIST ) ], ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=global_participant_id, audience_ids=None, target=None, include_archived=True, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == [] @pytest.mark.db def test_find_by_criteria_search_complex( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: adele_audience = create_model( Audience, name="Adele Audience", target=AudienceTarget.ADS, ) create_model( Audience, name="Adele test", vendor_id=adele_audience.vendor_id, subaccount_id=adele_audience.subaccount_id, target=AudienceTarget.ADS, ) create_model( Audience, name="Audience 1234", vendor_id=adele_audience.vendor_id, subaccount_id=adele_audience.subaccount_id, target=AudienceTarget.EMAIL, ) result = repository.find_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[adele_audience.vendor_id], ad_target_subaccount_ids=[adele_audience.subaccount_id], email_target_vendor_ids=[adele_audience.vendor_id], email_target_subaccount_ids=[adele_audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search="Adele Audience", global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, include_archived=False, ), order_by="createdAt.desc", limit=10, offset=0, identity_id=identity_id, ) assert result == IsList(adele_audience) @pytest.mark.db def test_count_by_criteria( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: audience = create_model(Audience) create_model(Audience) total = repository.count_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[audience.vendor_id], ad_target_subaccount_ids=[audience.subaccount_id], email_target_vendor_ids=[audience.vendor_id], email_target_subaccount_ids=[audience.subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, target=None, audience_ids=None, ), identity_id=identity_id, ) assert total == 1 @pytest.mark.db def test_count_by_criteria_different_access_to_vendors_in_target_without_text_target_access( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: email_audience = create_model(Audience, target=AudienceTarget.EMAIL) ad_audience = create_model(Audience, target=AudienceTarget.ADS) create_model(Audience) text_audience = create_model(Audience, target=AudienceTarget.TEXT) total = repository.count_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[ad_audience.vendor_id], ad_target_subaccount_ids=[ad_audience.subaccount_id], email_target_vendor_ids=[email_audience.vendor_id], email_target_subaccount_ids=[email_audience.subaccount_id], text_target_vendor_ids=[text_audience.vendor_id], text_target_subaccount_ids=[text_audience.subaccount_id], search=None, global_participant_id=None, custom_list_id=None, target=None, audience_ids=None, ), identity_id=identity_id, ) assert total == 2 @pytest.mark.db def test_count_by_criteria_different_access_to_vendors_in_target_with_text_target_access( self, repository: AudienceRepository, create_model: CreateModel, enable_features: EnableFeatures, identity_id: str, ) -> None: email_audience = create_model(Audience, target=AudienceTarget.EMAIL) ad_audience = create_model(Audience, target=AudienceTarget.ADS) create_model(Audience) text_audience = create_model(Audience, target=AudienceTarget.TEXT) with enable_features([AUDIENCE_SHOW_SMS_CAMPAIGNS_PAGE]): total = repository.count_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[ad_audience.vendor_id], ad_target_subaccount_ids=[ad_audience.subaccount_id], email_target_vendor_ids=[email_audience.vendor_id], email_target_subaccount_ids=[email_audience.subaccount_id], text_target_vendor_ids=[text_audience.vendor_id], text_target_subaccount_ids=[text_audience.subaccount_id], search=None, global_participant_id=None, custom_list_id=None, target=None, audience_ids=None, ), identity_id=identity_id, ) assert total == 3 @pytest.mark.db def test_count_by_criteria_include_archived( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 create_model(Audience, vendor_id=vendor_id, subaccount_id=subaccount_id) create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, archived_at=timezone.now(), ) total = repository.count_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=None, include_archived=True, ), identity_id=identity_id, ) assert total == 2 @pytest.mark.db def test_count_by_criteria_target( self, repository: AudienceRepository, create_model: CreateModel, identity_id: str, ) -> None: vendor_id = 1 subaccount_id = 0 create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, target=AudienceTarget.ADS, ) create_model( Audience, vendor_id=vendor_id, subaccount_id=subaccount_id, target=AudienceTarget.EMAIL, ) total = repository.count_by_criteria( AudiencesQueryCriteria( ad_target_vendor_ids=[vendor_id], ad_target_subaccount_ids=[subaccount_id], email_target_vendor_ids=[vendor_id], email_target_subaccount_ids=[subaccount_id], text_target_vendor_ids=[], text_target_subaccount_ids=[], search=None, global_participant_id=None, custom_list_id=None, audience_ids=None, target=AudienceTarget.EMAIL, include_archived=True, ), identity_id=identity_id, ) assert total == 1