from unittest import mock import pytest from fansifter_common.auth.account import Account, AccountAccess from fansifter_common.auth.exceptions import PermissionDenied from dmp.ad_accounts.enums import AdAccountPlatform from dmp.ad_accounts.models import AdAccountDbt from dmp.app_connections.enums import AppConnectionStatus from dmp.meta.dtos import MetaAdAccountOverview from dmp.meta.enums import MetaUserConnectionStatus from dmp.meta.handlers.get_ad_accounts_v2 import ( GetMetaAdAccountsV2Handler, GetMetaAdAccountsV2Request, ) from dmp.meta.models import ( MetaAdAccount, MetaAdReportingConnection, MetaUserAdAccount, MetaUserConnection, ) from tests.unit.faker import FakerTyped from tests.unit.types import CreateModel, CreateReportingModel 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, ) -> GetMetaAdAccountsV2Request: return GetMetaAdAccountsV2Request( identity_id=identity_id, vendor_id=vendor_id, subaccount_id=subaccount_id, limit=limit or GetMetaAdAccountsV2Request.DEFAULT_LIMIT, offset=offset or GetMetaAdAccountsV2Request.DEFAULT_OFFSET, order_by=order_by if order_by is not None else GetMetaAdAccountsV2Request.DEFAULT_ORDER_BY, ) class TestGetMetaAdAccountsV2Handler: @pytest.mark.db def test_returns_empty_when_no_accounts( self, handler: GetMetaAdAccountsV2Handler, 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: GetMetaAdAccountsV2Handler, create_model: CreateModel, identity_id: str, account: Account, ) -> None: ad_account = create_model(MetaAdAccount, business_account_picture_key=None) conn = create_model(MetaAdReportingConnection, ad_accounts=[ad_account]) user_ad_account = create_model( MetaUserAdAccount, 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 == [ MetaAdAccountOverview( id=ad_account.id, user_ad_account_id=user_ad_account.id, external_id=ad_account.external_id, name=ad_account.name or "", business_account_name=ad_account.business_account_name, business_account_picture=ad_account.business_account_picture, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, is_self_connected=False, user_connection_status=MetaUserConnectionStatus.NOT_CONNECTED, user_self_connected=False, ad_reporting_connection_status=conn.status, ad_reporting_self_connected=False, campaigns_count=ad_account.campaigns_count, ) ] @pytest.mark.db def test_limit_restricts_items_but_total_reflects_full_count( self, handler: GetMetaAdAccountsV2Handler, 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(MetaAdAccount, business_account_picture_key=None) conn = create_model(MetaAdReportingConnection, ad_accounts=[ad_account]) create_model( MetaUserAdAccount, 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: GetMetaAdAccountsV2Handler, 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 name in ("aaa", "mmm", "zzz"): ad_account = create_model( MetaAdAccount, name=name, business_account_picture_key=None ) conn = create_model(MetaAdReportingConnection, ad_accounts=[ad_account]) create_model( MetaUserAdAccount, 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: GetMetaAdAccountsV2Handler, 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(MetaAdAccount, business_account_picture_key=None) conn_1 = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_1]) create_model( MetaUserAdAccount, 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(MetaAdAccount, business_account_picture_key=None) conn_2 = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_2]) create_model( MetaUserAdAccount, 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 @pytest.mark.db def test_subaccount_id_filter_returns_only_matching_accounts( self, handler: GetMetaAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: account_1 = Account(vendor_id=0, subaccount_id=1) account_2 = Account(vendor_id=0, subaccount_id=2) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account_1, account_2] ) identity_id = fake.uuid4_string() ad_account_1 = create_model(MetaAdAccount, business_account_picture_key=None) conn_1 = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_1]) create_model( MetaUserAdAccount, 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(MetaAdAccount, business_account_picture_key=None) conn_2 = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_2]) create_model( MetaUserAdAccount, 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: GetMetaAdAccountsV2Handler, 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: GetMetaAdAccountsV2Handler, 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: GetMetaAdAccountsV2Handler, 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(MetaAdAccount, business_account_picture_key=None) conn = create_model(MetaAdReportingConnection, ad_accounts=[ad_account]) create_model( MetaUserAdAccount, 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_total_matches_items_count_with_pagination( self, handler: GetMetaAdAccountsV2Handler, 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(MetaAdAccount, business_account_picture_key=None) conn = create_model(MetaAdReportingConnection, ad_accounts=[ad_account]) create_model( MetaUserAdAccount, ad_account_id=ad_account.id, identity_id=conn.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) page1 = handler.handle(_make_request(identity_id, limit=2, offset=0)) page2 = handler.handle(_make_request(identity_id, limit=2, offset=2)) page3 = handler.handle(_make_request(identity_id, limit=2, offset=4)) assert page1.total == 5 assert page2.total == 5 assert page3.total == 5 assert len(list(page1.items)) == 2 assert len(list(page2.items)) == 2 assert len(list(page3.items)) == 1 assert ( len(list(page1.items)) + len(list(page2.items)) + len(list(page3.items)) == page1.total ) @pytest.mark.db def test_count_returns_zero_and_items_query_skipped_when_no_accounts( self, handler: GetMetaAdAccountsV2Handler, 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) == [] # only the count query should have been executed, not the items query assert mock_execute.call_count == 1 @pytest.mark.db def test_account_labeled_with_different_label_is_excluded( self, handler: GetMetaAdAccountsV2Handler, 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(MetaAdAccount, business_account_picture_key=None) 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=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: GetMetaAdAccountsV2Handler, 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(MetaAdAccount, business_account_picture_key=None) ad_account_b = create_model(MetaAdAccount, business_account_picture_key=None) create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account_a, ad_account_b], ) create_model( MetaUserAdAccount, 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( MetaUserAdAccount, 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_campaigns_count_desc_returns_accounts_in_descending_order( self, handler: GetMetaAdAccountsV2Handler, 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( MetaAdAccount, business_account_picture_key=None, campaigns_count=1 ) conn_low = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_low]) create_model( MetaUserAdAccount, 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( MetaAdAccount, business_account_picture_key=None, campaigns_count=99 ) conn_high = create_model( MetaAdReportingConnection, ad_accounts=[ad_account_high] ) create_model( MetaUserAdAccount, 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: GetMetaAdAccountsV2Handler, create_model: CreateModel, identity_id: str, account: Account, ) -> None: ad_account_labeled = create_model( MetaAdAccount, business_account_picture_key=None ) conn = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_labeled]) create_model( MetaUserAdAccount, 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( MetaAdAccount, business_account_picture_key=None ) create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[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_order_by_campaigns_count_desc_corrected_after_dbt_fill( self, handler: GetMetaAdAccountsV2Handler, create_model: CreateModel, create_reporting_model: CreateReportingModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: """SQL sorts by DB campaigns_count; fill may increase it from DBT — verify re-sort.""" account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() # DB count is low but DBT count is high — should end up first after fill ad_account_low_db = create_model( MetaAdAccount, business_account_picture_key=None, campaigns_count=1 ) conn_low = create_model( MetaAdReportingConnection, ad_accounts=[ad_account_low_db] ) create_model( MetaUserAdAccount, ad_account_id=ad_account_low_db.id, identity_id=conn_low.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) create_reporting_model( AdAccountDbt, id=ad_account_low_db.external_id, platform=AdAccountPlatform.META, campaigns_count=99, ) # DB count is high but DBT count is lower — max keeps DB value, should end up second ad_account_high_db = create_model( MetaAdAccount, business_account_picture_key=None, campaigns_count=10 ) conn_high = create_model( MetaAdReportingConnection, ad_accounts=[ad_account_high_db] ) create_model( MetaUserAdAccount, ad_account_id=ad_account_high_db.id, identity_id=conn_high.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) create_reporting_model( AdAccountDbt, id=ad_account_high_db.external_id, platform=AdAccountPlatform.META, campaigns_count=5, ) 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_low_db.id assert items[0].campaigns_count == 99 assert items[1].id == ad_account_high_db.id assert items[1].campaigns_count == 10 @pytest.mark.db def test_order_by_campaigns_count_asc_corrected_after_dbt_fill( self, handler: GetMetaAdAccountsV2Handler, create_model: CreateModel, create_reporting_model: CreateReportingModel, 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() # DB count is high but DBT count is even higher — should end up last on ASC ad_account_a = create_model( MetaAdAccount, business_account_picture_key=None, campaigns_count=10 ) conn_a = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_a]) create_model( MetaUserAdAccount, ad_account_id=ad_account_a.id, identity_id=conn_a.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) create_reporting_model( AdAccountDbt, id=ad_account_a.external_id, platform=AdAccountPlatform.META, campaigns_count=99, ) # DB count is low and DBT doesn't raise it — should end up first on ASC ad_account_b = create_model( MetaAdAccount, business_account_picture_key=None, campaigns_count=1 ) conn_b = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_b]) create_model( MetaUserAdAccount, ad_account_id=ad_account_b.id, identity_id=conn_b.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle( _make_request(identity_id, order_by=["campaignsCount.asc"]) ) items = list(response.items) assert len(items) == 2 assert items[0].id == ad_account_b.id assert items[0].campaigns_count == 1 assert items[1].id == ad_account_a.id assert items[1].campaigns_count == 99 @pytest.mark.db def test_order_by_name_desc_corrected_after_dbt_fill( self, handler: GetMetaAdAccountsV2Handler, create_model: CreateModel, create_reporting_model: CreateReportingModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: """When name equals external_id, fill replaces it with DBT name — verify re-sort.""" account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() # name == external_id so fill will replace it with "Aardvark" ad_account_a = create_model(MetaAdAccount, business_account_picture_key=None) ad_account_a.name = ad_account_a.external_id conn_a = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_a]) create_model( MetaUserAdAccount, ad_account_id=ad_account_a.id, identity_id=conn_a.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) create_reporting_model( AdAccountDbt, id=ad_account_a.external_id, platform=AdAccountPlatform.META, name="Aardvark", ) # name is already set and won't be replaced by fill ad_account_b = create_model( MetaAdAccount, name="Zebra", business_account_picture_key=None ) conn_b = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_b]) create_model( MetaUserAdAccount, ad_account_id=ad_account_b.id, identity_id=conn_b.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_b.id # "Zebra" > "Aardvark" assert items[1].id == ad_account_a.id # "Aardvark" @pytest.mark.db def test_order_by_business_account_name_desc_corrected_after_dbt_fill( self, handler: GetMetaAdAccountsV2Handler, create_model: CreateModel, create_reporting_model: CreateReportingModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, ) -> None: """NULL business_account_name sorts first on DESC in SQL but fill replaces it — verify re-sort.""" account = Account(vendor_id=1, subaccount_id=0) auth_service_mock.authorize_for_permission.return_value = AccountAccess( accounts=[account] ) identity_id = fake.uuid4_string() # business_account_name is None — SQL DESC puts it first; fill sets it to "Apple" ad_account_a = create_model( MetaAdAccount, business_account_name=None, business_account_picture_key=None, ) conn_a = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_a]) create_model( MetaUserAdAccount, ad_account_id=ad_account_a.id, identity_id=conn_a.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) create_reporting_model( AdAccountDbt, id=ad_account_a.external_id, platform=AdAccountPlatform.META, business_account_name="Apple", ) # business_account_name is already set and fill won't override it ad_account_b = create_model( MetaAdAccount, business_account_name="Zebra", business_account_picture_key=None, ) conn_b = create_model(MetaAdReportingConnection, ad_accounts=[ad_account_b]) create_model( MetaUserAdAccount, ad_account_id=ad_account_b.id, identity_id=conn_b.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle( _make_request(identity_id, order_by=["businessAccountName.desc"]) ) items = list(response.items) assert len(items) == 2 assert items[0].id == ad_account_b.id # "Zebra" > "Apple" assert items[1].id == ad_account_a.id # "Apple" @pytest.mark.db def test_self_sharing_account_without_label_is_returned( self, handler: GetMetaAdAccountsV2Handler, create_model: CreateModel, identity_id: str, ) -> None: ad_account = create_model(MetaAdAccount, business_account_picture_key=None) create_model( MetaUserConnection, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account], ) response = handler.handle(_make_request(identity_id)) assert response.total == 1 items = list(response.items) assert len(items) == 1 assert items[0] == MetaAdAccountOverview( id=ad_account.id, user_ad_account_id=None, external_id=ad_account.external_id, name=ad_account.name or "", business_account_name=ad_account.business_account_name, business_account_picture=ad_account.business_account_picture, vendor_id=None, subaccount_id=None, is_self_connected=True, user_connection_status=MetaUserConnectionStatus.TOS_NOT_ACCEPTED, user_self_connected=True, ad_reporting_connection_status=AppConnectionStatus.NOT_CONNECTED, ad_reporting_self_connected=False, campaigns_count=ad_account.campaigns_count, )