import pathlib from unittest import mock import pytest from faker import Faker from freezegun import freeze_time from app.adapters.ows_dmp import AudienceSharePlatform, AudienceShareStatus from app.exceptions import HandlerError from app.handler import ShareAudienceHandler, ShareAudienceRequest from app.strategies import GoogleAudienceShareStrategy from tests.unit.adapters.google.factories import UserDataOfflineJobFactory from tests.unit.factories import AudienceShareFactory, GoogleAudienceShareFactory def test_share_google_audience_share_failed( handler: ShareAudienceHandler, ows_dmp_client_mock: mock.MagicMock, s3_client_mock: mock.MagicMock, audience_file: pathlib.Path, mock_download_fileobj: None, ) -> None: audience_share = AudienceShareFactory.build( platform=AudienceSharePlatform.GOOGLE, ) ows_dmp_client_mock.update_audience_share.return_value = audience_share ows_dmp_client_mock.get_google_audience_share.side_effect = Exception("error") with pytest.raises(HandlerError) as exc_info: handler.handle( ShareAudienceRequest( bucket="test", key=f"shares/{audience_share.id}/{audience_share.id}_audience.csv", ) ) assert exc_info.value.code == "GET_GOOGLE_AUDIENCE_SHARE_ERROR" ows_dmp_client_mock.update_audience_share.assert_has_calls( [ mock.call( audience_share.id, status=AudienceShareStatus.FAILED, processed_fans=None, ) ] ) def test_share_google_audience_decrypt_failed( handler: ShareAudienceHandler, ows_dmp_client_mock: mock.MagicMock, s3_client_mock: mock.MagicMock, kms_client_mock: mock.MagicMock, mock_download_fileobj: None, ) -> None: audience_share = AudienceShareFactory.build(platform=AudienceSharePlatform.GOOGLE) share_data = GoogleAudienceShareFactory.build() ows_dmp_client_mock.update_audience_share.return_value = audience_share ows_dmp_client_mock.get_google_audience_share.return_value = share_data kms_client_mock.decrypt.side_effect = Exception("error") with pytest.raises(HandlerError) as exc_info: handler.handle( ShareAudienceRequest( bucket="test", key=f"shares/{audience_share.id}/{audience_share.id}_audience.csv", ) ) assert exc_info.value.code == "DECRYPT_USER_ACCESS_TOKEN_ERROR" ows_dmp_client_mock.update_audience_share.assert_has_calls( [ mock.call( audience_share.id, status=AudienceShareStatus.FAILED, processed_fans=None, ) ] ) @freeze_time("2023-11-03T12:09:52.902512") def test_share_google_audience_upload_users_failed( handler: ShareAudienceHandler, ows_dmp_client_mock: mock.MagicMock, s3_client_mock: mock.MagicMock, kms_client_mock: mock.MagicMock, google_client_mock: mock.MagicMock, mock_download_fileobj: None, ) -> None: audience_share = AudienceShareFactory.build( platform=AudienceSharePlatform.GOOGLE, ) share_data = GoogleAudienceShareFactory.build() ows_dmp_client_mock.update_audience_share.return_value = audience_share ows_dmp_client_mock.get_google_audience_share.return_value = share_data kms_client_mock.decrypt.return_value = share_data.user_access_token google_client_mock.create_offline_user_data_job.side_effect = Exception("error") with pytest.raises(HandlerError) as exc_info: handler.handle( ShareAudienceRequest( bucket="test", key=f"shares/{audience_share.id}/{audience_share.id}_audience.csv", ) ) assert exc_info.value.code == "PROCESS_AUDIENCE_ERROR" ows_dmp_client_mock.update_audience_share.assert_has_calls( [ mock.call( audience_share.id, status=AudienceShareStatus.PROCESSING, processed_fans=None, ), mock.call( audience_share.id, status=AudienceShareStatus.FAILED, processed_fans=None, ), ] ) google_client_mock.add_to_offline_user_data_job.assert_not_called() google_client_mock.start_offline_user_data_job.assert_not_called() @freeze_time("2023-11-03T12:09:52.902512") def test_share_google_audience( handler: ShareAudienceHandler, ows_dmp_client_mock: mock.MagicMock, s3_client_mock: mock.MagicMock, kms_client_mock: mock.MagicMock, google_client_mock: mock.MagicMock, mock_download_fileobj_google: None, google_strategy: GoogleAudienceShareStrategy, faker: Faker, ) -> None: # Set chunk size to 3 to test multiple batches login_customer_id = str(faker.pyint()) google_strategy.upload_chunksize = 3 audience_share = AudienceShareFactory.build( platform=AudienceSharePlatform.GOOGLE, ) google_audience_share = GoogleAudienceShareFactory.build( fan_count=10, parentAdAccountExternalId=login_customer_id ) ows_dmp_client_mock.update_audience_share.return_value = audience_share ows_dmp_client_mock.get_google_audience_share.return_value = google_audience_share kms_client_mock.decrypt.return_value = google_audience_share.user_access_token offline_job = UserDataOfflineJobFactory.build() google_client_mock.create_offline_user_data_job.return_value = offline_job handler.handle( ShareAudienceRequest( bucket="test", key=f"shares/{audience_share.id}/{audience_share.id}_audience.csv", ) ) ows_dmp_client_mock.update_audience_share.assert_has_calls( [ mock.call( audience_share.id, status=AudienceShareStatus.PROCESSING, processed_fans=None, ), mock.call( audience_share.id, status=AudienceShareStatus.PROCESSING, processed_fans=3, ), mock.call( audience_share.id, status=AudienceShareStatus.PROCESSING, processed_fans=6, ), mock.call( audience_share.id, status=AudienceShareStatus.PROCESSING, processed_fans=9, ), mock.call( audience_share.id, status=AudienceShareStatus.COMPLETED, processed_fans=10, ), ] ) google_client_mock.create_offline_user_data_job.assert_has_calls( [ mock.call( access_token=google_audience_share.user_access_token, customer_id=google_audience_share.ad_account_external_id, user_list_resource=google_audience_share.user_list_resource, login_customer_id=login_customer_id, ) ] ) google_client_mock.add_to_offline_user_data_job.assert_has_calls( [ mock.call( access_token=google_audience_share.user_access_token, offline_job_resource=offline_job.resource_name, payload=[ ["hash1", ""], ["hash2", "hash2"], ["hash3", "hash3"], ], login_customer_id=login_customer_id, ), mock.call( access_token=google_audience_share.user_access_token, offline_job_resource=offline_job.resource_name, payload=[ ["hash4", "hash4"], ["hash5", "hash5"], ["hash6", "hash6"], ], login_customer_id=login_customer_id, ), mock.call( access_token=google_audience_share.user_access_token, offline_job_resource=offline_job.resource_name, payload=[ ["hash7", "hash7"], ["hash8", "hash8"], ["hash9", "hash9"], ], login_customer_id=login_customer_id, ), mock.call( access_token=google_audience_share.user_access_token, offline_job_resource=offline_job.resource_name, payload=[ ["", "hash10"], ], login_customer_id=login_customer_id, ), ], ) google_client_mock.start_offline_user_data_job.assert_has_calls( [ mock.call( access_token=google_audience_share.user_access_token, offline_job_resource=offline_job.resource_name, login_customer_id=login_customer_id, ) ] )