"""Unit tests for TikTok and Douyin tasks workflow.""" import os from unittest import mock from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import patch import freezegun import pytest from feed_ingestion.flows.tiktok import config from feed_ingestion.flows.tiktok import tasks _date = '2021-01-03' @pytest.fixture def reports_info_theorchard(): """Return downloaded_files.""" return { 'TikTok': { 'drop_bucket': 'prod-orcd-ftp', 'drop_path': ( 'ftp/data_feed/tiktok/' 'TikTok_Daily_Usage_Report_20210103.tsv.zip'), 'archive_path': ( 'TikTok/archives/2021-01-03/' 'TikTok_Daily_Usage_Report_20210103.tsv.zip'), 'processed_path': ( 'TikTok/processed/2021-01-03/' 'TikTok_Daily_Usage_Report_20210103.tsv.gz'), 'feed_name': 'tiktok_theorchard_TikTok', 'temp_staging_raw_table': 'temp_staging_raw_theorchard_TikTok_20210103'}, 'Douyin': { 'drop_bucket': 'prod-orcd-ftp', 'drop_path': ( 'ftp/data_feed/tiktok/' 'Douyin_Daily_Usage_Report_20210103.tsv.zip'), 'archive_path': ( 'TikTok/archives/2021-01-03/' 'Douyin_Daily_Usage_Report_20210103.tsv.zip'), 'processed_path': ( 'TikTok/processed/2021-01-03/' 'Douyin_Daily_Usage_Report_20210103.tsv.gz'), 'feed_name': 'tiktok_theorchard_Douyin', 'temp_staging_raw_table': 'temp_staging_raw_theorchard_Douyin_20210103'} } @pytest.fixture def reports_info_theorchard_sme_s3(): """Return reports_info when theorchard uses the SME S3 bucket.""" return { 'TikTok': { 'drop_bucket': 'sme-ca-prod-partners', 'drop_path': ( 'tiktok/in/orchard/daily_trends/' 'TikTok_Daily_Usage_Report_20210103.tsv.zip'), 'archive_path': ( 'TikTok/archives/2021-01-03/' 'TikTok_Daily_Usage_Report_20210103.tsv.zip'), 'processed_path': ( 'TikTok/processed/2021-01-03/' 'TikTok_Daily_Usage_Report_20210103.tsv.gz'), 'feed_name': 'tiktok_theorchard_TikTok', 'temp_staging_raw_table': 'temp_staging_raw_theorchard_TikTok_20210103'}, 'Douyin': { 'drop_bucket': 'sme-ca-prod-partners', 'drop_path': ( 'tiktok/in/orchard/daily_trends/' 'Douyin_Daily_Usage_Report_20210103.tsv.zip'), 'archive_path': ( 'TikTok/archives/2021-01-03/' 'Douyin_Daily_Usage_Report_20210103.tsv.zip'), 'processed_path': ( 'TikTok/processed/2021-01-03/' 'Douyin_Daily_Usage_Report_20210103.tsv.gz'), 'feed_name': 'tiktok_theorchard_Douyin', 'temp_staging_raw_table': 'temp_staging_raw_theorchard_Douyin_20210103'} } @pytest.fixture def expected_bootstrap_response_theorchard(reports_info_theorchard): """Response for bootstrap task.""" return { 'feed_name': 'tiktok_theorchard', 'date': '2021-01-03', 'reports_info': reports_info_theorchard } @pytest.fixture def expected_bootstrap_response_theorchard_sme_s3( reports_info_theorchard_sme_s3): """Response for bootstrap task when theorchard uses the SME S3 bucket.""" return { 'feed_name': 'tiktok_theorchard', 'date': '2021-01-03', 'reports_info': reports_info_theorchard_sme_s3 } @pytest.fixture def reports_info_sme(): """Return downloaded_files.""" return { 'TikTok': { 'drop_bucket': 'sme-ca-prod-partners', 'drop_path': ( 'tiktok/in/sme/daily_trends/PLY3_D_ROW_20210103.txt'), 'archive_path': ( 'TikTok/archives_sme/2021-01-03/' 'PLY3_D_ROW_20210103.txt'), 'processed_path': ( 'TikTok/archives_sme/2021-01-03/PLY3_D_ROW_20210103.txt'), 'feed_name': 'tiktok_sme_TikTok', 'temp_staging_raw_table': 'temp_staging_raw_sme_TikTok_20210103'}, 'Douyin': { 'drop_bucket': 'sme-ca-prod-partners', 'drop_path': ( 'tiktok/in/sme/daily_trends/PLY3_D_CN_20210103.txt'), 'archive_path': ( 'TikTok/archives_sme/2021-01-03/' 'PLY3_D_CN_20210103.txt'), 'processed_path': ( 'TikTok/archives_sme/2021-01-03/PLY3_D_CN_20210103.txt'), 'feed_name': 'tiktok_sme_Douyin', 'temp_staging_raw_table': 'temp_staging_raw_sme_Douyin_20210103'} } @pytest.fixture def expected_bootstrap_response_sme(reports_info_sme): """Response for bootstrap task.""" return { 'feed_name': 'tiktok_sme', 'date': '2021-01-03', 'reports_info': reports_info_sme } @pytest.fixture def reports_info_smejpintl(): """Return downloaded_files.""" return { 'TikTok': { 'drop_bucket': 'sme-ca-prod-partners', 'drop_path': 'tiktok/in/smej/ww/PLY3_D_ROW_20210103.txt', 'archive_path': os.path.join( 'TikTok', 'archives_smejpintl', '2021-01-03', 'PLY3_D_ROW_20210103.txt' ), 'processed_path': os.path.join( 'TikTok', 'archives_smejpintl', '2021-01-03', 'PLY3_D_ROW_20210103.txt' ), 'feed_name': 'tiktok_smejpintl_TikTok', 'temp_staging_raw_table': 'temp_staging_raw_smejpintl_TikTok_20210103' } } @pytest.fixture def expected_bootstrap_response_smejpintl(reports_info_smejpintl): """Response for bootstrap task.""" return { 'feed_name': 'tiktok_smejpintl', 'date': '2021-01-03', 'reports_info': reports_info_smejpintl } @pytest.fixture def reports_info_smejp(): """Return downloaded_files.""" return { 'TikTok': { 'drop_bucket': 'sme-ca-prod-partners', 'drop_path': ( 'tiktok/in/smej/japan/PLY3_D_ROW_20210103.txt'), 'archive_path': ( 'TikTok/archives_smejp/2021-01-03/' 'PLY3_D_ROW_20210103.txt'), 'processed_path': ( 'TikTok/archives_smejp/2021-01-03/PLY3_D_ROW_20210103.txt'), 'feed_name': 'tiktok_smejp_TikTok', 'temp_staging_raw_table': 'temp_staging_raw_smejp_TikTok_20210103'} } @pytest.fixture def expected_bootstrap_response_smejp(reports_info_smejp): """Response for bootstrap task.""" return { 'feed_name': 'tiktok_smejp', 'date': '2021-01-03', 'reports_info': reports_info_smejp } @pytest.fixture def mock_get_overall_status(): """Yield get overall status.""" overall_status_path = ( 'feed_ingestion.flows.tiktok.tasks.garcon_feed_status.' 'get_overall_status') with patch(overall_status_path) as overall_status: yield overall_status @pytest.fixture def mock_set_overall_status(): """Yield overall status.""" overall_status_path = ( 'feed_ingestion.flows.tiktok.tasks.set_overall_status_enhanced') with patch(overall_status_path) as overall_status: yield overall_status @pytest.fixture def mock_delete_status(): """Yield overall status.""" overall_status_path = ( 'feed_ingestion.flows.tiktok.tasks.garcon_feed_status.' 'delete_status') with patch(overall_status_path) as overall_status: yield overall_status def test_bootstrap_theorchard( expected_bootstrap_response_theorchard, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports=None, reload=False, feed_name='tiktok_theorchard', licensor='theorchard', use_sme_s3=None) assert result == expected_bootstrap_response_theorchard mock_delete_status.assert_not_called() def test_bootstrap_with_report_theorchard( reports_info_theorchard, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports='TikTok', feed_name='tiktok_theorchard', reload=False, licensor='theorchard', use_sme_s3=None) assert result['reports_info'] == { 'TikTok': reports_info_theorchard['TikTok']} mock_delete_status.assert_not_called() def test_bootstrap_with_reload_theorchard( expected_bootstrap_response_theorchard, mock_delete_status): """Test bootstrap task.""" licensor = 'theorchard' result = tasks.bootstrap( MagicMock(), _date, reports=None, reload='True', feed_name=F'tiktok_{licensor}', licensor=licensor, use_sme_s3=None) assert result == expected_bootstrap_response_theorchard mock_delete_status.assert_has_calls([ call(f'{config.feed_name}_{licensor}_{report}', _date) for report in config.reports]) def test_bootstrap_theorchard_sme_s3( expected_bootstrap_response_theorchard_sme_s3, mock_delete_status): """Test bootstrap task when theorchard uses the SME S3 bucket.""" result = tasks.bootstrap( MagicMock(), _date, reports=None, reload=False, feed_name='tiktok_theorchard', licensor='theorchard', use_sme_s3='True') assert result == expected_bootstrap_response_theorchard_sme_s3 mock_delete_status.assert_not_called() def test_bootstrap_with_report_theorchard_sme_s3( reports_info_theorchard_sme_s3, mock_delete_status): """Test bootstrap task with single report when theorchard uses SME S3.""" result = tasks.bootstrap( MagicMock(), _date, reports='TikTok', feed_name='tiktok_theorchard', reload=False, licensor='theorchard', use_sme_s3='True') assert result['reports_info'] == { 'TikTok': reports_info_theorchard_sme_s3['TikTok']} mock_delete_status.assert_not_called() def test_bootstrap_with_reload_theorchard_sme_s3( expected_bootstrap_response_theorchard_sme_s3, mock_delete_status): """Test bootstrap task with reload when theorchard uses SME S3.""" licensor = 'theorchard' result = tasks.bootstrap( MagicMock(), _date, reports=None, reload='True', feed_name=f'tiktok_{licensor}', licensor=licensor, use_sme_s3='True') assert result == expected_bootstrap_response_theorchard_sme_s3 mock_delete_status.assert_has_calls([ call(f'{config.feed_name}_{licensor}_{report}', _date) for report in config.reports]) def test_bootstrap_sme( expected_bootstrap_response_sme, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports=None, reload=False, feed_name='tiktok_sme', licensor='sme', use_sme_s3=None) assert result == expected_bootstrap_response_sme mock_delete_status.assert_not_called() def test_bootstrap_with_report_sme( reports_info_sme, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports='TikTok', feed_name='tiktok_sme', reload=False, licensor='sme', use_sme_s3=None) assert result['reports_info'] == { 'TikTok': reports_info_sme['TikTok']} mock_delete_status.assert_not_called() def test_bootstrap_with_reload_sme( expected_bootstrap_response_sme, mock_delete_status): """Test bootstrap task.""" licensor = 'sme' result = tasks.bootstrap( MagicMock(), _date, reports=None, reload='True', feed_name=F'tiktok_{licensor}', licensor=licensor, use_sme_s3=None) assert result == expected_bootstrap_response_sme mock_delete_status.assert_has_calls([ call(f'{config.feed_name}_{licensor}_{report}', _date) for report in config.reports]) def test_bootstrap_smejp( expected_bootstrap_response_smejp, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports=None, reload=False, feed_name='tiktok_smejp', licensor='smejp', use_sme_s3=None) assert result == expected_bootstrap_response_smejp mock_delete_status.assert_not_called() def test_bootstrap_with_report_smejp( reports_info_smejp, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports='TikTok', feed_name='tiktok_smejp', reload=False, licensor='smejp', use_sme_s3=None) assert result['reports_info'] == { 'TikTok': reports_info_smejp['TikTok']} mock_delete_status.assert_not_called() def test_bootstrap_with_reload_smejp( expected_bootstrap_response_smejp, mock_delete_status): """Test bootstrap task.""" supported_reports = ['TikTok'] licensor = 'smejp' result = tasks.bootstrap( MagicMock(), _date, reports=','.join(supported_reports), reload='True', feed_name=f'tiktok_{licensor}', licensor=licensor, use_sme_s3=None ) assert result == expected_bootstrap_response_smejp mock_delete_status.assert_has_calls([ call(f'{config.feed_name}_{licensor}_{report}', _date) for report in supported_reports]) def test_bootstrap_smejpintl( expected_bootstrap_response_smejpintl, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports=None, reload=False, feed_name='tiktok_smejpintl', licensor='smejpintl', use_sme_s3=None ) assert result == expected_bootstrap_response_smejpintl mock_delete_status.assert_not_called() def test_bootstrap_with_report_smejpintl( reports_info_smejpintl, mock_delete_status): """Test bootstrap task.""" result = tasks.bootstrap( MagicMock(), _date, reports='TikTok', feed_name='tiktok_smejpintl', reload=False, licensor='smejpintl', use_sme_s3=None) assert result['reports_info'] == { 'TikTok': reports_info_smejpintl['TikTok']} mock_delete_status.assert_not_called() def test_bootstrap_with_reload_smejpintl( expected_bootstrap_response_smejpintl, mock_delete_status): """Test bootstrap task.""" supported_reports = ['TikTok'] licensor = 'smejpintl' result = tasks.bootstrap( MagicMock(), _date, reports=','.join(supported_reports), reload='True', feed_name=f'tiktok_{licensor}', licensor=licensor, use_sme_s3=None ) assert result == expected_bootstrap_response_smejpintl mock_delete_status.assert_has_calls([ call(f'{config.feed_name}_{licensor}_{report}', _date) for report in supported_reports]) @pytest.fixture def mock_task_status(): """Yield task status.""" task_status_path = 'feed_ingestion.tasks.task_status' with patch(task_status_path) as task_status: task_status.is_completed_task.return_value = False task_status.mark_completed_task = MagicMock() task_status.get_values = MagicMock() yield task_status @pytest.fixture def mock_task_status_tiktok(): """Yield task status.""" task_status_path = 'feed_ingestion.flows.tiktok.tasks.task_status' with patch(task_status_path) as task_status: task_status.is_completed_task.return_value = False task_status.mark_completed_task = MagicMock() task_status.get_values = MagicMock() yield task_status @pytest.fixture def mock_s3_tasks(): """Yield overall status.""" path = 'feed_ingestion.flows.tiktok.tasks.s3_tasks' with patch(path) as mock_s3: yield mock_s3 @pytest.fixture def mock_convert_zip_to_gzip_on_s3(): """Yield overall status.""" path = 'feed_ingestion.flows.tiktok.tasks.convert_zip_to_gzip_on_s3' with patch(path) as mock_convert: yield mock_convert def test_grab_drop_files_file_is_not_available_theorchard( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_theorchard): """Test test_grab_drop_files when a file are not available.""" mock_s3_tasks.copy_file.return_value = {} for report, info in reports_info_theorchard.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'theorchard', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result == {'stop': True} mock_set_overall_status.assert_called_with( 'feed_name', _date, 'NOT_AVAILABLE', mock.ANY) mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.assert_not_called() def test_grab_drop_files_file_is_available_theorchard( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_theorchard, mock_convert_zip_to_gzip_on_s3): """Test test_grab_drop_files when a file is available.""" mock_s3_tasks.copy_file.return_value = { 'TikTok_Daily_Usage_Report_20210103.tsv.zip': True, 'Douyin_Daily_Usage_Report_20210103.tsv.zip': True } for report, info in reports_info_theorchard.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'theorchard', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result != {'stop': True} mock_convert_zip_to_gzip_on_s3.call_count = len(reports_info_theorchard) mock_set_overall_status.assert_not_called() mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.assert_not_called() def test_grab_drop_files_file_is_not_available_theorchard_sme_s3( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_theorchard_sme_s3): """Test grab_drop_files when file is not available via SME S3.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = {} for report, info in reports_info_theorchard_sme_s3.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'theorchard', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result == {'stop': True} mock_set_overall_status.assert_called_with( 'feed_name', _date, 'NOT_AVAILABLE', mock.ANY) mock_s3_tasks.copy_file.assert_not_called() def test_grab_drop_files_file_is_available_theorchard_sme_s3( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_theorchard_sme_s3, mock_convert_zip_to_gzip_on_s3): """Test grab_drop_files when file is available via SME S3.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = { 'TikTok_Daily_Usage_Report_20210103.tsv.zip': True, 'Douyin_Daily_Usage_Report_20210103.tsv.zip': True } for report, info in reports_info_theorchard_sme_s3.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'theorchard', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result != {'stop': True} mock_convert_zip_to_gzip_on_s3.call_count = len( reports_info_theorchard_sme_s3) mock_set_overall_status.assert_not_called() mock_s3_tasks.copy_file.assert_not_called() def test_grab_drop_files_file_is_not_available_sme( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_sme): """Test test_grab_drop_files when a file are not available.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = {} for report, info in reports_info_sme.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'sme', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result == {'stop': True} mock_set_overall_status.assert_called_with( 'feed_name', _date, 'NOT_AVAILABLE', mock.ANY) mock_s3_tasks.copy_file.assert_not_called() def test_grab_drop_files_file_is_available_sme( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_sme, mock_convert_zip_to_gzip_on_s3): """Test test_grab_drop_files when a file is available.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = { 'PLY3_D_ROW_20210103.txt': True, 'PLY3_D_CN_20210103.txt': True } for report, info in reports_info_sme.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'sme', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result != {'stop': True} mock_convert_zip_to_gzip_on_s3.call_count = len(reports_info_sme) mock_set_overall_status.assert_not_called() mock_s3_tasks.copy_file.assert_not_called() def test_grab_drop_files_file_is_available_smejp( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_smejp, mock_convert_zip_to_gzip_on_s3): """Test grab_drop_files when a file is available for smejp.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = { 'PLY3_D_ROW_20210103.txt': True } for report, info in reports_info_smejp.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'smejp', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result != {'stop': True} mock_convert_zip_to_gzip_on_s3.call_count = len(reports_info_smejp) mock_set_overall_status.assert_not_called() mock_s3_tasks.copy_file.assert_not_called() def test_grab_drop_files_file_is_not_available_smejpintl( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_smejpintl): """Test grab_drop_files when a file is not available for smejpintl.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = {} for report, info in reports_info_smejpintl.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'smejpintl', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result == {'stop': True} mock_set_overall_status.assert_called_with( 'feed_name', _date, 'NOT_AVAILABLE', mock.ANY) mock_s3_tasks.copy_file.assert_not_called() def test_check_available_reports_theorchard( mock_set_overall_status, reports_info_theorchard, mock_task_status_tiktok): """Test check_available_reports when all reports are downloaded.""" mock_task_status_tiktok.is_completed_task.return_value = True tasks.check_available_reports( MagicMock(), _date, reports_info_theorchard, 'tiktok_theorchard') mock_set_overall_status.assert_called_once_with( 'tiktok_theorchard', _date, 'DOWNLOADED', mock.ANY) def test_check_available_reports_not_available_theorchard( mock_set_overall_status, reports_info_theorchard, mock_task_status_tiktok): """Test check_available_reports when reports are not downloaded.""" mock_task_status_tiktok.is_completed_task.return_value = False result = tasks.check_available_reports( MagicMock(), _date, reports_info_theorchard, 'tiktok_theorchard') assert result == \ {'stop': True, 'missing_reports': {'Douyin': 'ftp/data_feed/tiktok' '/Douyin_Daily_Usage_Report_20210103.tsv.zip', 'TikTok': 'ftp/data_feed/tiktok' '/TikTok_Daily_Usage_Report_20210103.tsv.zip'}} mock_set_overall_status.assert_not_called() def test_check_available_reports_some_not_available_theorchard( mock_set_overall_status, reports_info_theorchard, mock_task_status_tiktok): """Test check_available_reports when reports are not downloaded.""" def is_completed_task_side_effect(*args): return False if args[0] == 'tiktok_theorchard_TikTok' else True mock_task_status_tiktok.is_completed_task.side_effect = \ is_completed_task_side_effect result = tasks.check_available_reports( MagicMock(), _date, reports_info_theorchard, 'tiktok_theorchard') assert result == \ {'stop': True, 'missing_reports': {'TikTok': 'ftp/data_feed/tiktok' '/TikTok_Daily_Usage_Report_20210103.tsv.zip'}} mock_set_overall_status.assert_not_called() def test_grab_drop_files_file_is_available_smejpintl( mock_task_status, mock_set_overall_status, mock_s3_tasks, reports_info_smejpintl, mock_convert_zip_to_gzip_on_s3): """Test grab_drop_files when a file is available for smejpintl.""" mock_s3_tasks.copy_file_from_sme_s3_to_theocrhard.return_value = { 'PLY3_D_ROW_20210103.txt': True } for report, info in reports_info_smejpintl.items(): result = tasks.grab_drop_files( MagicMock(), 'feed_name', 'smejpintl', _date, info['drop_bucket'], info['drop_path'], info['archive_path'], info['processed_path']) assert result != {'stop': True} mock_convert_zip_to_gzip_on_s3.call_count = len(reports_info_smejpintl) mock_set_overall_status.assert_not_called() mock_s3_tasks.copy_file.assert_not_called() @freezegun.freeze_time('2025-10-18') def test_set_status_to_ingested( mock_set_overall_status, mock_get_overall_status): """Test set_status_to_ingested when all reports are ingested.""" mock_get_overall_status.return_value = 'POPULATED_RAW_TABLE' tasks.set_status_to_ingested( MagicMock(), _date, 'theorchard', 'tiktok_theorchard', 'TikTok') mock_set_overall_status.assert_called_with( feed_name='tiktok_theorchard', date=_date, status='INGESTED', activity=mock.ANY) def test_set_status_to_ingested_reports_are_not_ingested( mock_get_overall_status, mock_set_overall_status): """Test set_status_to_ingested when reports are not ingested.""" mock_get_overall_status.return_value = 'NOT_AVAILABLE' tasks.set_status_to_ingested( MagicMock(), _date, 'theorchard', 'tiktok_theorchard', 'TikTok') mock_set_overall_status.assert_not_called() @patch('feed_ingestion.flows.tiktok.tasks.TikTok') @patch('feed_ingestion.flows.tiktok.tasks.merge_configs') @patch('feed_ingestion.flows.tiktok.tasks.get_sf_config') def test_remap_fingerprint_isrc_product_code_theorchard_enabled( mock_get_sf_config, mock_merge_configs, mock_tiktok_class): """Test remap_fingerprint_isrc_product_code for theorchard. Tests with flag 'True'. """ # Setup mocks mock_activity = MagicMock() sfdb_params = {'db': 'test_db', 'schema': 'test_schema'} mock_sf_config = {'db': 'test_db', 'schema': 'test_schema'} mock_get_sf_config.return_value = mock_sf_config mock_merge_configs.return_value = mock_sf_config mock_executor = MagicMock() mock_tiktok_class.return_value.__enter__.return_value = mock_executor # Execute tasks.remap_fingerprint_isrc_product_code( activity=mock_activity, date=_date, feed_name='tiktok_theorchard', sfdb_params=sfdb_params, licensor='theorchard' ) # Verify mock_get_sf_config.assert_called_once_with(config.secrets_path) mock_merge_configs.assert_called_once_with( mock_sf_config, sfdb_params) mock_tiktok_class.assert_called_once_with(mock_sf_config) mock_executor.update_staging_raw_table.assert_called_once_with( staging_raw_table=config.staging_raw_table, date=_date ) @patch('feed_ingestion.flows.tiktok.tasks.TikTok') @patch('feed_ingestion.flows.tiktok.tasks.merge_configs') @patch('feed_ingestion.flows.tiktok.tasks.get_sf_config') def test_remap_fingerprint_isrc_product_code_sme_licensor( mock_get_sf_config, mock_merge_configs, mock_tiktok_class): """Test remap_fingerprint_isrc_product_code for non-theorchard licensor.""" # Setup mocks mock_activity = MagicMock() sfdb_params = {'db': 'test_db', 'schema': 'test_schema'} # Execute with sme licensor tasks.remap_fingerprint_isrc_product_code( activity=mock_activity, date=_date, feed_name='tiktok_sme', sfdb_params=sfdb_params, licensor='sme' ) # Verify - should skip execution because licensor is not theorchard mock_activity.logger.info.assert_called_once_with( 'Skipping remap_fingerprint_isrc_product_code task' ' as it is not required.') mock_get_sf_config.assert_not_called() mock_merge_configs.assert_not_called() mock_tiktok_class.assert_not_called() @patch('feed_ingestion.flows.tiktok.tasks.TikTok') @patch('feed_ingestion.flows.tiktok.tasks.merge_configs') @patch('feed_ingestion.flows.tiktok.tasks.get_sf_config') def test_remap_fingerprint_isrc_product_code_altafonte_licensor( mock_get_sf_config, mock_merge_configs, mock_tiktok_class): """Test remap_fingerprint_isrc_product_code for altafonte licensor.""" # Setup mocks mock_activity = MagicMock() sfdb_params = {'db': 'test_db', 'schema': 'test_schema'} # Execute with altafonte licensor (even with flag enabled, should skip) tasks.remap_fingerprint_isrc_product_code( activity=mock_activity, date=_date, feed_name='tiktok_altafonte', sfdb_params=sfdb_params, licensor='altafonte', ) # Verify - should skip execution because licensor is not theorchard mock_activity.logger.info.assert_called_once_with( 'Skipping remap_fingerprint_isrc_product_code task' ' as it is not required.') mock_get_sf_config.assert_not_called() mock_merge_configs.assert_not_called() mock_tiktok_class.assert_not_called()