import pytest from dirty_equals import IsList from dmp.adapters.features import AUDIENCE_ALLOW_AUDIENCE_SHARING_TO_ALL_AD_ACCOUNTS from dmp.audiences.dtos import AudienceAdAccount from dmp.audiences.enums import AudienceSharePlatform from dmp.audiences.models import Audience from dmp.audiences.repositories import AudienceAdAccountRepository from dmp.google.models import ( GoogleAdAccount, GoogleUserAdAccount, GoogleUserConnection, GoogleUserConnectionAdAccount, ) from dmp.meta.models import ( MetaAdAccount, MetaUserAdAccount, MetaUserConnection, ) from dmp.tiktok.models import TikTokAdAccount, TikTokUserAdAccount, TikTokUserConnection from tests.unit.faker import FakerTyped from tests.unit.types import CreateModel, EnableFeatures class TestAudienceAdAccountRepository: @pytest.mark.db def test_find_for_audience_sharing( self, repository: AudienceAdAccountRepository, create_model: CreateModel, fake: FakerTyped, identity_id: str, ) -> None: audience = create_model(Audience) user_3 = fake.uuid4_string() # meta models setup meta_ad_account_1 = create_model(MetaAdAccount) meta_ad_account_2 = create_model(MetaAdAccount) meta_ad_account_3 = create_model(MetaAdAccount) meta_ad_account_3.user_custom_audience_tos_accepted = True create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account_1], identity_id=identity_id, ) create_model( MetaUserConnection, is_valid=False, ad_accounts=[meta_ad_account_1] ) create_model( MetaUserConnection, is_valid=False, ad_accounts=[meta_ad_account_1, meta_ad_account_2], ) create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account_3], identity_id=user_3, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model(MetaUserAdAccount, ad_account_id=meta_ad_account_2.id) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_3.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=user_3, ) # tiktok models setup tiktok_ad_account_1 = create_model(TikTokAdAccount) tiktok_ad_account_2 = create_model(TikTokAdAccount) tiktok_ad_account_3 = create_model(TikTokAdAccount) create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account_1], identity_id=identity_id, ) create_model( TikTokUserConnection, is_valid=False, ad_accounts=[tiktok_ad_account_1] ) create_model( TikTokUserConnection, is_valid=False, ad_accounts=[tiktok_ad_account_1, tiktok_ad_account_2], ) create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account_3], identity_id=user_3, ) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model(TikTokUserAdAccount, ad_account_id=tiktok_ad_account_2.id) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account_3.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=user_3, ) # google models setup google_ad_account_1 = create_model(GoogleAdAccount) google_ad_account_2 = create_model(GoogleAdAccount) google_ad_account_3 = create_model(GoogleAdAccount) create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_1) ], identity_id=identity_id, ) create_model( GoogleUserConnection, is_valid=False, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_1) ], ) create_model( GoogleUserConnection, is_valid=False, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_1), GoogleUserConnectionAdAccount(ad_account=google_ad_account_2), ], ) create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_3) ], identity_id=user_3, ) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model(GoogleUserAdAccount, ad_account_id=google_ad_account_2.id) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account_3.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=user_3, ) result = repository.find_for_audience_sharing( audience_id=audience.id, identity_id=identity_id ) assert len(result) == 6 assert result == IsList( AudienceAdAccount( id=meta_ad_account_1.id, external_id=meta_ad_account_1.external_id, platform=AudienceSharePlatform.META, name=meta_ad_account_1.name or meta_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=False, ), AudienceAdAccount( id=meta_ad_account_3.id, external_id=meta_ad_account_3.external_id, platform=AudienceSharePlatform.META, name=meta_ad_account_3.name or meta_ad_account_3.external_id, is_connected=True, is_self_connected=False, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=tiktok_ad_account_1.id, external_id=tiktok_ad_account_1.external_id, platform=AudienceSharePlatform.TIKTOK, name=tiktok_ad_account_1.name or tiktok_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=tiktok_ad_account_3.id, external_id=tiktok_ad_account_3.external_id, platform=AudienceSharePlatform.TIKTOK, name=tiktok_ad_account_3.name or tiktok_ad_account_3.external_id, is_connected=True, is_self_connected=False, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=google_ad_account_1.id, external_id=google_ad_account_1.external_id, platform=AudienceSharePlatform.GOOGLE, name=google_ad_account_1.name or google_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=google_ad_account_3.id, external_id=google_ad_account_3.external_id, platform=AudienceSharePlatform.GOOGLE, name=google_ad_account_3.name or google_ad_account_3.external_id, is_connected=True, is_self_connected=False, custom_audience_tos_accepted=True, ), check_order=False, ) @pytest.mark.db def test_find_for_audience_sharing_when_sharing_to_all_ad_accounts_not_allowed( self, repository: AudienceAdAccountRepository, create_model: CreateModel, fake: FakerTyped, identity_id: str, ) -> None: vendor_id_1 = 1000 vendor_id_2 = 2000 identity_id_other = fake.uuid4_string() audience = create_model(Audience, vendor_id=vendor_id_1, subaccount_id=0) # meta models setup meta_ad_account_1 = create_model(MetaAdAccount) meta_ad_account_1.user_custom_audience_tos_accepted = True meta_ad_account_2 = create_model(MetaAdAccount) meta_ad_account_2.user_custom_audience_tos_accepted = True create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account_1], identity_id=identity_id, ) create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account_2], identity_id=identity_id_other, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_2.id, vendor_id=vendor_id_2, subaccount_id=0, identity_id=identity_id_other, ) # tiktok models setup tiktok_ad_account_1 = create_model(TikTokAdAccount) tiktok_ad_account_2 = create_model(TikTokAdAccount) create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account_1], identity_id=identity_id, ) create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account_2], identity_id=identity_id_other, ) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account_2.id, vendor_id=vendor_id_2, subaccount_id=0, identity_id=identity_id_other, ) # google models setup google_ad_account_1 = create_model(GoogleAdAccount) google_ad_account_2 = create_model(GoogleAdAccount) create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_1) ], identity_id=identity_id, ) create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_2) ], identity_id=identity_id_other, ) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account_2.id, vendor_id=vendor_id_2, subaccount_id=0, identity_id=identity_id_other, ) result = repository.find_for_audience_sharing( audience_id=audience.id, identity_id=identity_id ) assert len(result) == 3 assert result == IsList( AudienceAdAccount( id=meta_ad_account_1.id, external_id=meta_ad_account_1.external_id, platform=AudienceSharePlatform.META, name=meta_ad_account_1.name or meta_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=tiktok_ad_account_1.id, external_id=tiktok_ad_account_1.external_id, platform=AudienceSharePlatform.TIKTOK, name=tiktok_ad_account_1.name or tiktok_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=google_ad_account_1.id, external_id=google_ad_account_1.external_id, platform=AudienceSharePlatform.GOOGLE, name=google_ad_account_1.name or google_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), check_order=False, ) @pytest.mark.db def test_find_for_audience_sharing_when_sharing_to_all_ad_accounts_allowed( self, repository: AudienceAdAccountRepository, create_model: CreateModel, fake: FakerTyped, identity_id: str, enable_features: EnableFeatures, ) -> None: vendor_id_1 = 1000 vendor_id_2 = 2000 vendor_id_3 = 3000 identity_id_other = fake.uuid4_string() audience = create_model(Audience, vendor_id=vendor_id_1, subaccount_id=0) # meta models setup meta_ad_account_1 = create_model(MetaAdAccount) meta_ad_account_1.user_custom_audience_tos_accepted = True meta_ad_account_2 = create_model(MetaAdAccount) meta_ad_account_2.user_custom_audience_tos_accepted = True meta_ad_account_3 = create_model(MetaAdAccount) meta_ad_account_3.user_custom_audience_tos_accepted = True create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account_1], identity_id=identity_id, ) create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account_2, meta_ad_account_3], identity_id=identity_id_other, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_2.id, vendor_id=vendor_id_2, subaccount_id=0, identity_id=identity_id_other, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account_3.id, vendor_id=vendor_id_3, subaccount_id=0, identity_id=identity_id_other, ) # tiktok models setup tiktok_ad_account_1 = create_model(TikTokAdAccount) tiktok_ad_account_2 = create_model(TikTokAdAccount) create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account_1], identity_id=identity_id, ) create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account_2], identity_id=identity_id_other, ) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account_2.id, vendor_id=vendor_id_2, subaccount_id=0, identity_id=identity_id_other, ) # google models setup google_ad_account_1 = create_model(GoogleAdAccount) google_ad_account_2 = create_model(GoogleAdAccount) create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_1) ], identity_id=identity_id, ) create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account_2) ], identity_id=identity_id_other, ) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account_1.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account_2.id, vendor_id=vendor_id_2, subaccount_id=0, identity_id=identity_id_other, ) with enable_features([AUDIENCE_ALLOW_AUDIENCE_SHARING_TO_ALL_AD_ACCOUNTS]): result = repository.find_for_audience_sharing( audience_id=audience.id, identity_id=identity_id, vendor_ids=[vendor_id_1, vendor_id_2], ) assert len(result) == 6 assert result == IsList( AudienceAdAccount( id=meta_ad_account_1.id, external_id=meta_ad_account_1.external_id, platform=AudienceSharePlatform.META, name=meta_ad_account_1.name or meta_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=meta_ad_account_2.id, external_id=meta_ad_account_2.external_id, platform=AudienceSharePlatform.META, name=meta_ad_account_2.name or meta_ad_account_2.external_id, is_connected=True, is_self_connected=False, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=tiktok_ad_account_1.id, external_id=tiktok_ad_account_1.external_id, platform=AudienceSharePlatform.TIKTOK, name=tiktok_ad_account_1.name or tiktok_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=tiktok_ad_account_2.id, external_id=tiktok_ad_account_2.external_id, platform=AudienceSharePlatform.TIKTOK, name=tiktok_ad_account_2.name or tiktok_ad_account_2.external_id, is_connected=True, is_self_connected=False, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=google_ad_account_1.id, external_id=google_ad_account_1.external_id, platform=AudienceSharePlatform.GOOGLE, name=google_ad_account_1.name or google_ad_account_1.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=google_ad_account_2.id, external_id=google_ad_account_2.external_id, platform=AudienceSharePlatform.GOOGLE, name=google_ad_account_2.name or google_ad_account_2.external_id, is_connected=True, is_self_connected=False, custom_audience_tos_accepted=True, ), check_order=False, ) @pytest.mark.db def test_find_by_accounts_for_audience_sharing_empty( self, repository: AudienceAdAccountRepository, fake: FakerTyped ) -> None: result = repository.find_for_audience_sharing( audience_id=fake.uuid4_string(), identity_id=fake.uuid4_string() ) assert result == [] @pytest.mark.db def test_find_for_audience_sharing_empty_name( self, repository: AudienceAdAccountRepository, create_model: CreateModel, identity_id: str, enable_features: EnableFeatures, ) -> None: vendor_id_1 = 1000 vendor_id_2 = 2000 audience = create_model(Audience, vendor_id=vendor_id_1, subaccount_id=0) # meta models setup meta_ad_account = create_model(MetaAdAccount, name="") meta_ad_account.user_custom_audience_tos_accepted = True create_model( MetaUserConnection, is_valid=True, ad_accounts=[meta_ad_account], identity_id=identity_id, ) create_model( MetaUserAdAccount, ad_account_id=meta_ad_account.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) # tiktok models setup tiktok_ad_account = create_model(TikTokAdAccount, name="") create_model( TikTokUserConnection, is_valid=True, ad_accounts=[tiktok_ad_account], identity_id=identity_id, ) create_model( TikTokUserAdAccount, ad_account_id=tiktok_ad_account.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) # google models setup google_ad_account = create_model(GoogleAdAccount, name="") create_model( GoogleUserConnection, is_valid=True, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=google_ad_account) ], identity_id=identity_id, ) create_model( GoogleUserAdAccount, ad_account_id=google_ad_account.id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, identity_id=identity_id, ) with enable_features([AUDIENCE_ALLOW_AUDIENCE_SHARING_TO_ALL_AD_ACCOUNTS]): result = repository.find_for_audience_sharing( audience_id=audience.id, identity_id=identity_id, vendor_ids=[vendor_id_1, vendor_id_2], ) assert len(result) == 3 assert result == IsList( AudienceAdAccount( id=meta_ad_account.id, external_id=meta_ad_account.external_id, platform=AudienceSharePlatform.META, name=meta_ad_account.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=tiktok_ad_account.id, external_id=tiktok_ad_account.external_id, platform=AudienceSharePlatform.TIKTOK, name=tiktok_ad_account.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), AudienceAdAccount( id=google_ad_account.id, external_id=google_ad_account.external_id, platform=AudienceSharePlatform.GOOGLE, name=google_ad_account.external_id, is_connected=True, is_self_connected=True, custom_audience_tos_accepted=True, ), check_order=False, )