from unittest import mock import httpx import pytest from fansifter_common.auth.account import Account, AccountAccess from dmp.adapters.features import AUDIENCE_ALLOW_AUDIENCE_SHARING_TO_ALL_AD_ACCOUNTS from dmp.adapters.meta.exceptions import FacebookClientError from dmp.adapters.meta.models import CustomAudience, DebugToken, User from dmp.audiences.enums import ( AudienceExportJustification, AudienceExportStatus, AudienceSharePlatform, ) from dmp.audiences.exceptions import ( AudienceShareActiveConnectionNotFoundError, EmptyAudienceError, ) from dmp.audiences.models import Audience, AudienceSnapshot from dmp.config import Settings from dmp.meta.exceptions import InvalidMetaAdAccountId, MetaAdAccountPermissionsError from dmp.meta.handlers import ShareMetaAudienceHandler, ShareMetaAudienceRequest from dmp.meta.models import MetaAdAccount, MetaUserAdAccount, MetaUserConnection from tests.unit.faker import FakerTyped from tests.unit.types import BuildModel, CreateModel, EnableFeatures class TestShareMetaAudienceHandler: @pytest.fixture(autouse=True) def mock_facebook_client( self, facebook_client_mock: mock.MagicMock, build_model: BuildModel, settings: Settings, ) -> None: facebook_client_mock.debug_token.return_value = DebugToken( is_valid=True, scopes=settings.meta_user_required_scopes ) facebook_client_mock.get_user.return_value = build_model(User) @pytest.mark.db def test_share_audience( self, handler: ShareMetaAudienceHandler, create_model: CreateModel, identity_id: str, facebook_client_mock: mock.MagicMock, ) -> None: justification = AudienceExportJustification.OTHER reason_notes = "test" ad_account = create_model(MetaAdAccount) audience = create_model(Audience) snapshot = create_model(AudienceSnapshot, audience=audience) user_connection = create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account], ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=user_connection.identity_id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) facebook_client_mock.create_custom_audience.return_value = CustomAudience( id="123" ) meta_audience = handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=ad_account.id, justification=justification, reason_notes=reason_notes, ) ) assert meta_audience.audience_id == snapshot.audience_id assert meta_audience.snapshot_id == snapshot.id assert meta_audience.ad_account_id == ad_account.id assert meta_audience.name == f"[FS] {audience.name}" assert meta_audience.share.status == AudienceExportStatus.PENDING assert meta_audience.share.platform == AudienceSharePlatform.META assert meta_audience.share.justification == justification assert meta_audience.share.reason_notes == reason_notes assert meta_audience.share.created_by == identity_id @pytest.mark.db def test_share_audience_no_active_user_connection( self, handler: ShareMetaAudienceHandler, create_model: CreateModel, identity_id: str, ) -> None: ad_account = create_model(MetaAdAccount) audience = create_model(Audience) snapshot = create_model(AudienceSnapshot, audience=audience, fan_count=100) user_connection = create_model( MetaUserConnection, is_valid=False, identity_id=identity_id, ad_accounts=[ad_account], ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=user_connection.identity_id, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) with pytest.raises(AudienceShareActiveConnectionNotFoundError) as exc_info: handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=ad_account.id, justification=AudienceExportJustification.OTHER, reason_notes="test", ) ) assert exc_info.value.code == "audience_share_active_connection_not_found" assert ( exc_info.value.message == "Active connection for Audience share not found" ) @pytest.mark.db def test_share_audience_no_user_ad_account( self, handler: ShareMetaAudienceHandler, create_model: CreateModel, identity_id: str, fake: FakerTyped, ) -> None: snapshot = create_model(AudienceSnapshot) with pytest.raises(InvalidMetaAdAccountId): handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=fake.uuid4_string(), justification=AudienceExportJustification.OTHER, reason_notes="test", ) ) @pytest.mark.db def test_share_audience_without_snapshot( self, handler: ShareMetaAudienceHandler, create_model: CreateModel, identity_id: str, fake: FakerTyped, ) -> None: ad_account_id = fake.pystr() audience = create_model(Audience) with pytest.raises(EmptyAudienceError): handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=audience.id, ad_account_id=ad_account_id, justification=AudienceExportJustification.OTHER, reason_notes="test", ) ) @pytest.mark.db def test_share_audience_with_zero_fans( self, handler: ShareMetaAudienceHandler, create_model: CreateModel, identity_id: str, fake: FakerTyped, ) -> None: ad_account_id = fake.pystr() snapshot = create_model(AudienceSnapshot, fan_count=0) with pytest.raises(EmptyAudienceError): handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=ad_account_id, justification=AudienceExportJustification.OTHER, reason_notes="test", ) ) @pytest.mark.db def test_share_audience_create_custom_audience_error( self, handler: ShareMetaAudienceHandler, facebook_client_mock: mock.MagicMock, create_model: CreateModel, identity_id: str, account: Account, ) -> None: justification = AudienceExportJustification.OTHER reason_notes = "test" ad_account = create_model(MetaAdAccount) audience = create_model( Audience, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id ) snapshot = create_model(AudienceSnapshot, audience=audience) user_connection = create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account], ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=user_connection.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) facebook_client_mock.create_custom_audience.side_effect = FacebookClientError with pytest.raises(FacebookClientError): handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=ad_account.id, justification=justification, reason_notes=reason_notes, ) ) @pytest.mark.db def test_share_audience_create_custom_audience_permissions_error_user_connection( self, handler: ShareMetaAudienceHandler, facebook_client_mock: mock.MagicMock, create_model: CreateModel, identity_id: str, account: Account, ) -> None: justification = AudienceExportJustification.OTHER reason_notes = "test" ad_account = create_model(MetaAdAccount) audience = create_model( Audience, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id ) snapshot = create_model(AudienceSnapshot, audience=audience) user_connection = create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account], ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) facebook_client_mock.create_custom_audience.side_effect = FacebookClientError( response=httpx.Response( status_code=400, json={ "error": { "code": 200, } }, ) ) facebook_client_mock.get_user_ad_accounts.return_value = [] with pytest.raises(MetaAdAccountPermissionsError): handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=ad_account.id, justification=justification, reason_notes=reason_notes, ) ) assert user_connection.ad_accounts == [] @pytest.mark.db def test_share_audience_create_custom_audience_permissions_error_shared_connection( self, handler: ShareMetaAudienceHandler, facebook_client_mock: mock.MagicMock, create_model: CreateModel, fake: FakerTyped, account: Account, ) -> None: identity_id_1 = fake.uuid4_string() identity_id_2 = fake.uuid4_string() justification = AudienceExportJustification.OTHER reason_notes = "test" ad_account = create_model(MetaAdAccount) audience = create_model( Audience, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id ) snapshot = create_model(AudienceSnapshot, audience=audience, fan_count=100) user_connection = create_model( MetaUserConnection, is_valid=False, identity_id=identity_id_1, ad_accounts=[ad_account], ) create_model( MetaUserConnection, is_valid=True, identity_id=identity_id_2, ad_accounts=[ad_account], ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=identity_id_1, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=identity_id_2, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) facebook_client_mock.create_custom_audience.side_effect = FacebookClientError( response=httpx.Response( status_code=400, json={ "error": { "code": 200, } }, ) ) facebook_client_mock.get_user_ad_accounts.return_value = [] with pytest.raises(MetaAdAccountPermissionsError): handler.handle( ShareMetaAudienceRequest( identity_id=identity_id_1, audience_id=snapshot.audience_id, ad_account_id=ad_account.id, justification=justification, reason_notes=reason_notes, ) ) assert user_connection.ad_accounts == [] @pytest.mark.db def test_share_audience_to_an_ad_account_of_different_vendor( self, handler: ShareMetaAudienceHandler, create_model: CreateModel, identity_id: str, facebook_client_mock: mock.MagicMock, auth_service_mock: mock.MagicMock, enable_features: EnableFeatures, ) -> None: justification = AudienceExportJustification.OTHER reason_notes = "test" audience_vendor_id = 1000 ad_account_vendor_id = 2000 ad_account = create_model(MetaAdAccount) audience = create_model(Audience, vendor_id=audience_vendor_id, subaccount_id=0) snapshot = create_model(AudienceSnapshot, audience=audience, fan_count=1001) user_connection = create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account], ) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=user_connection.identity_id, vendor_id=ad_account_vendor_id, subaccount_id=0, ) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[ Account(vendor_id=audience_vendor_id, subaccount_id=0), Account(vendor_id=ad_account_vendor_id, subaccount_id=0), ], ) facebook_client_mock.create_custom_audience.return_value = CustomAudience( id="123" ) with enable_features([AUDIENCE_ALLOW_AUDIENCE_SHARING_TO_ALL_AD_ACCOUNTS]): meta_audience = handler.handle( ShareMetaAudienceRequest( identity_id=identity_id, audience_id=snapshot.audience_id, ad_account_id=ad_account.id, justification=justification, reason_notes=reason_notes, ) ) assert meta_audience.audience_id == snapshot.audience_id assert meta_audience.snapshot_id == snapshot.id assert meta_audience.ad_account_id == ad_account.id assert meta_audience.name == f"[FS] {audience.name}" assert meta_audience.share.status == AudienceExportStatus.PENDING assert meta_audience.share.platform == AudienceSharePlatform.META assert meta_audience.share.justification == justification assert meta_audience.share.reason_notes == reason_notes assert meta_audience.share.created_by == identity_id