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.tiktok.dtos import TikTokAdAccountOverview from dmp.tiktok.enums import TikTokUserConnectionStatus from dmp.tiktok.handlers.get_ad_accounts_v2 import ( GetTikTokAdAccountsV2Handler, GetTikTokAdAccountsV2Request, ) from dmp.tiktok.models import ( TikTokAdAccount, TikTokAdReportingConnection, TikTokUserAdAccount, TikTokUserConnection, ) 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, ) -> GetTikTokAdAccountsV2Request: return GetTikTokAdAccountsV2Request( identity_id=identity_id, vendor_id=vendor_id, subaccount_id=subaccount_id, limit=limit or GetTikTokAdAccountsV2Request.DEFAULT_LIMIT, offset=offset or GetTikTokAdAccountsV2Request.DEFAULT_OFFSET, order_by=order_by if order_by is not None else GetTikTokAdAccountsV2Request.DEFAULT_ORDER_BY, ) class TestGetTikTokAdAccountsV2Handler: @pytest.mark.db def test_returns_empty_when_no_accounts( self, handler: GetTikTokAdAccountsV2Handler, 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: GetTikTokAdAccountsV2Handler, create_model: CreateModel, identity_id: str, account: Account, ) -> None: ad_account = create_model(TikTokAdAccount) conn = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) user_ad_account = create_model( TikTokUserAdAccount, 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 == [ TikTokAdAccountOverview( 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, business_center_name=ad_account.business_center_name, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, is_self_connected=False, user_connection_status=TikTokUserConnectionStatus.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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount) conn = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount) conn = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount) conn_1 = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_1], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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(TikTokAdAccount) conn_2 = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_2], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, 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: GetTikTokAdAccountsV2Handler, 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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount) conn = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, 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: GetTikTokAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, user_id: str, ) -> 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(TikTokAdAccount) create_model( TikTokUserConnection, user_id=user_id, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account], ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, create_model: CreateModel, auth_service_mock: mock.MagicMock, fake: FakerTyped, user_id: str, ) -> 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(TikTokAdAccount) ad_account_b = create_model(TikTokAdAccount) create_model( TikTokUserConnection, user_id=user_id, is_valid=True, identity_id=identity_id, ad_accounts=[ad_account_a, ad_account_b], ) create_model( TikTokUserAdAccount, 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( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount, name="A account") conn_a = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_a], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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(TikTokAdAccount, name="Z account") conn_z = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_z], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount, campaigns_count=1) conn_low = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_low], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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(TikTokAdAccount, campaigns_count=99) conn_high = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_high], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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: GetTikTokAdAccountsV2Handler, create_model: CreateModel, identity_id: str, account: Account, user_id: str, ) -> None: ad_account_labeled = create_model(TikTokAdAccount) conn = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_labeled], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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(TikTokAdAccount) create_model( TikTokUserConnection, user_id=user_id, 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_business_center_name_asc( self, handler: GetTikTokAdAccountsV2Handler, 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_z = create_model(TikTokAdAccount, business_center_name="Z Center") conn_z = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_z], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, ad_account_id=ad_account_z.id, identity_id=conn_z.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) ad_account_a = create_model(TikTokAdAccount, business_center_name="A Center") conn_a = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_a], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, ad_account_id=ad_account_a.id, identity_id=conn_a.identity_id, vendor_id=account.vendor_id, subaccount_id=account.subaccount_id, ) response = handler.handle( _make_request(identity_id, order_by=["businessCenterName.asc"]) ) items = list(response.items) assert len(items) == 2 assert items[0].id == ad_account_a.id assert items[1].id == ad_account_z.id @pytest.mark.db def test_order_by_business_center_name_desc( self, handler: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount, business_center_name="A Center") conn_a = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_a], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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(TikTokAdAccount, business_center_name="Z Center") conn_z = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_z], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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=["businessCenterName.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_corrected_after_dbt_fill( self, handler: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount, campaigns_count=1) conn_low = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_low_db], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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.TIKTOK, 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(TikTokAdAccount, campaigns_count=10) conn_high = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_high_db], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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.TIKTOK, 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: GetTikTokAdAccountsV2Handler, 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(TikTokAdAccount, campaigns_count=10) conn_a = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_a], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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.TIKTOK, campaigns_count=99, ) # DB count is low and DBT doesn't raise it — should end up first on ASC ad_account_b = create_model(TikTokAdAccount, campaigns_count=1) conn_b = create_model( TikTokAdReportingConnection, ad_accounts=[ad_account_b], status=AppConnectionStatus.CONNECTED, ) create_model( TikTokUserAdAccount, 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_self_sharing_account_without_label_is_returned( self, handler: GetTikTokAdAccountsV2Handler, create_model: CreateModel, identity_id: str, user_id: str, ) -> None: ad_account = create_model(TikTokAdAccount) conn = create_model( TikTokUserConnection, user_id=user_id, 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] == TikTokAdAccountOverview( 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, business_center_name=ad_account.business_center_name, vendor_id=None, subaccount_id=None, is_self_connected=True, user_connection_status=TikTokUserConnectionStatus.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, )