from unittest import mock import pytest from fansifter_common.auth.account import Account, AccountAccess from fansifter_common.auth.exceptions import PermissionDenied from dmp.app_connections.enums import AppConnectionStatus from dmp.google.dtos import GoogleAdAccountOverview from dmp.google.enums import GoogleUserConnectionStatus from dmp.google.handlers.get_ad_accounts_v2 import ( GetGoogleAdAccountsV2Handler, GetGoogleAdAccountsV2Request, ) from dmp.google.models import ( GoogleAdAccount, GoogleAdReportingConnection, GoogleUserAdAccount, GoogleUserConnection, GoogleUserConnectionAdAccount, ) from tests.unit.faker import FakerTyped from tests.unit.types import CreateModel def _make_request( identity_id: str, *, vendor_id: int | None = None, subaccount_id: int | None = None, limit: int | None = None, offset: int | None = None, order_by: list | None = None, ) -> GetGoogleAdAccountsV2Request: return GetGoogleAdAccountsV2Request( identity_id=identity_id, vendor_id=vendor_id, subaccount_id=subaccount_id, limit=limit or GetGoogleAdAccountsV2Request.DEFAULT_LIMIT, offset=offset or GetGoogleAdAccountsV2Request.DEFAULT_OFFSET, order_by=order_by if order_by is not None else GetGoogleAdAccountsV2Request.DEFAULT_ORDER_BY, ) class TestGetGoogleAdAccountsV2Handler: @pytest.mark.db def test_returns_empty_when_no_accounts( self, handler: GetGoogleAdAccountsV2Handler, identity_id: str, ) -> None: response = handler.handle(_make_request(identity_id)) assert response.total == 0 assert list(response.items) == [] @pytest.mark.db def test_returns_accounts_with_total( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, identity_id: str, account: Account, ) -> None: ad_account = create_model(GoogleAdAccount) conn = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) user_ad_account = create_model( GoogleUserAdAccount, ad_account_id=ad_account.id, identity_id=conn.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle(_make_request(identity_id)) assert response.total == 1 assert response.items == [ GoogleAdAccountOverview( id=ad_account.id, user_ad_account_id=user_ad_account.id, user_id=None, external_id=ad_account.external_id, name=ad_account.name or ad_account.external_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, is_self_connected=False, user_connection_status=GoogleUserConnectionStatus.NOT_CONNECTED, user_self_connected=False, ad_reporting_connection_status=AppConnectionStatus.CONNECTED, ad_reporting_self_connected=False, sharing_user_id=None, reporting_user_id=None, campaigns_count=ad_account.campaigns_count, ) ] @pytest.mark.db def test_limit_restricts_items_but_total_reflects_full_count( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() for _ in range(3): ad_account = create_model(GoogleAdAccount) conn = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account.id, identity_id=conn.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle(_make_request(identity_id, limit=2)) assert response.total == 3 assert len(list(response.items)) == 2 @pytest.mark.db def test_offset_skips_items( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() for _ in range(3): ad_account = create_model(GoogleAdAccount) conn = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account.id, identity_id=conn.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response_all = handler.handle(_make_request(identity_id)) response_offset = handler.handle(_make_request(identity_id, offset=1)) assert response_offset.total == 3 assert len(list(response_offset.items)) == 2 assert list(response_offset.items) == list(response_all.items)[1:] @pytest.mark.db def test_vendor_id_filter_returns_only_matching_accounts( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account_1 = Account(vendor_id=1, subaccount_id=0) account_2 = Account(vendor_id=2, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account_1, account_2] ) identity_id = fake.uuid4_string() ad_account_1 = create_model(GoogleAdAccount) conn_1 = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_1], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_1.id, identity_id=conn_1.identity_id, vendor_id=account_1.vendor_id, subaccount_id=account_1.subaccount_id, ) ad_account_2 = create_model(GoogleAdAccount) conn_2 = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_2], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_2.id, identity_id=conn_2.identity_id, vendor_id=account_2.vendor_id, subaccount_id=account_2.subaccount_id, ) response = handler.handle( _make_request( identity_id, vendor_id=account_1.vendor_id, subaccount_id=account_1.subaccount_id, ) ) assert response.total == 1 assert list(response.items)[0].id == ad_account_1.id def test_vendor_id_not_in_authorized_list_raises_permission_denied( self, handler: GetGoogleAdAccountsV2Handler, identity_id: str, auth_service_mock: mock.MagicMock, ) -> None: auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[Account(vendor_id=1, subaccount_id=0)] ) with pytest.raises(PermissionDenied): handler.handle(_make_request(identity_id, vendor_id=99, subaccount_id=0)) def test_subaccount_id_not_in_authorized_list_raises_permission_denied( self, handler: GetGoogleAdAccountsV2Handler, identity_id: str, auth_service_mock: mock.MagicMock, ) -> None: auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[Account(vendor_id=0, subaccount_id=1)] ) with pytest.raises(PermissionDenied): handler.handle(_make_request(identity_id, vendor_id=0, subaccount_id=99)) @pytest.mark.db def test_total_matches_items_count_without_pagination( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() for _ in range(5): ad_account = create_model(GoogleAdAccount) conn = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account.id, identity_id=conn.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle(_make_request(identity_id)) assert response.total == len(list(response.items)) @pytest.mark.db def test_count_returns_zero_and_items_query_skipped_when_no_accounts( self, handler: GetGoogleAdAccountsV2Handler, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() with mock.patch.object( handler.db.session, "execute", wraps=handler.db.session.execute ) as mock_execute: response = handler.handle(_make_request(identity_id)) assert response.total == 0 assert list(response.items) == [] assert mock_execute.call_count == 1 @pytest.mark.db def test_account_labeled_with_different_label_is_excluded( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: label_a = Account(vendor_id=1, subaccount_id=0) label_b = Account(vendor_id=2, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[label_a, label_b] ) identity_id = fake.uuid4_string() ad_account = create_model(GoogleAdAccount) create_model( GoogleUserConnection, is_valid=True, identity_id=identity_id, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=ad_account) ], ) create_model( GoogleUserAdAccount, ad_account_id=ad_account.id, identity_id=identity_id, vendor_id=label_a.vendor_id, subaccount_id=label_a.subaccount_id, ) response = handler.handle( _make_request( identity_id, vendor_id=label_b.vendor_id, subaccount_id=label_b.subaccount_id, ) ) assert response.total == 0 assert list(response.items) == [] @pytest.mark.db def test_only_matching_label_account_returned_when_two_labels_exist( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: label_a = Account(vendor_id=1, subaccount_id=0) label_b = Account(vendor_id=2, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[label_a, label_b] ) identity_id = fake.uuid4_string() ad_account_a = create_model(GoogleAdAccount) ad_account_b = create_model(GoogleAdAccount) create_model( GoogleUserConnection, is_valid=True, identity_id=identity_id, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=ad_account_a), GoogleUserConnectionAdAccount(ad_account=ad_account_b), ], ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_a.id, identity_id=identity_id, vendor_id=label_a.vendor_id, subaccount_id=label_a.subaccount_id, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_b.id, identity_id=identity_id, vendor_id=label_b.vendor_id, subaccount_id=label_b.subaccount_id, ) response = handler.handle( _make_request( identity_id, vendor_id=label_b.vendor_id, subaccount_id=label_b.subaccount_id, ) ) assert response.total == 1 assert list(response.items)[0].id == ad_account_b.id @pytest.mark.db def test_order_by_name_desc_returns_accounts_in_reverse_name_order( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() ad_account_a = create_model( GoogleAdAccount, name="A account", ) conn_a = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_a], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_a.id, identity_id=conn_a.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) ad_account_z = create_model( GoogleAdAccount, name="Z account", ) conn_z = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_z], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_z.id, identity_id=conn_z.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle(_make_request(identity_id, order_by=["name.desc"])) items = list(response.items) assert len(items) == 2 assert items[0].id == ad_account_z.id assert items[1].id == ad_account_a.id @pytest.mark.db def test_order_by_campaigns_count_desc_returns_accounts_in_descending_order( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() ad_account_low = create_model(GoogleAdAccount, campaigns_count=1) conn_low = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_low], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_low.id, identity_id=conn_low.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) ad_account_high = create_model(GoogleAdAccount, campaigns_count=99) conn_high = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_high], status=AppConnectionStatus.CONNECTED, ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_high.id, identity_id=conn_high.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle( _make_request(identity_id, order_by=["campaignsCount.desc"]) ) items = list(response.items) assert len(items) == 2 assert items[0].id == ad_account_high.id assert items[1].id == ad_account_low.id @pytest.mark.db def test_order_by_vendor_id_asc_puts_null_vendor_id_accounts_first( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, identity_id: str, account: Account, ) -> None: ad_account_labeled = create_model(GoogleAdAccount) conn = create_model( GoogleAdReportingConnection, ad_accounts=[ad_account_labeled], ) create_model( GoogleUserAdAccount, ad_account_id=ad_account_labeled.id, identity_id=conn.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) ad_account_unassigned = create_model(GoogleAdAccount) create_model( GoogleUserConnection, is_valid=True, identity_id=identity_id, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=ad_account_unassigned) ], ) response = handler.handle( _make_request(identity_id, order_by=["vendorId.asc.nullsFirst"]) ) items = list(response.items) assert len(items) == 2 assert items[0].id == ad_account_unassigned.id assert items[1].id == ad_account_labeled.id @pytest.mark.db def test_self_sharing_account_without_label_is_returned( self, handler: GetGoogleAdAccountsV2Handler, create_model: CreateModel, identity_id: str, ) -> None: ad_account = create_model(GoogleAdAccount) conn = create_model( GoogleUserConnection, is_valid=True, identity_id=identity_id, connection_ad_accounts=[ GoogleUserConnectionAdAccount(ad_account=ad_account) ], ) response = handler.handle(_make_request(identity_id)) assert response.total == 1 items = list(response.items) assert len(items) == 1 assert items[0] == GoogleAdAccountOverview( id=ad_account.id, user_ad_account_id=None, user_id=None, external_id=ad_account.external_id, name=ad_account.name or ad_account.external_id, vendor_id=None, subaccount_id=None, is_self_connected=True, user_connection_status=GoogleUserConnectionStatus.CONNECTED, user_self_connected=True, ad_reporting_connection_status=AppConnectionStatus.NOT_CONNECTED, ad_reporting_self_connected=False, sharing_user_id=conn.user_id, reporting_user_id=None, campaigns_count=ad_account.campaigns_count, )