"""Unit tests for tasks of YouTube Monthly Ingestion Workflow.""" import gzip from pathlib import Path from unittest.mock import call, MagicMock from unittest.mock import patch import pytest from feed_ingestion.flows.youtube_monthly import tasks SOURCE_FILES_DICT = {'files': [ { 'file_name': 'source_orchard.csv.gz', 'mcn_account': 'theorchardmusic', 'csv_header_lines': 2, 'file_size': 0, }, { 'file_name': 'source_dmgi.csv.gz', 'mcn_account': 'dmgi', 'csv_header_lines': 1, 'file_size': 0, }, ]} SOURCE_FILES_DICT_MULTIREPORT = {'files': [ { 'file_name': 'source_orchard_multireport.csv.gz', 'mcn_account': 'theorchardmusic', 'file_size': 0, 'subreports': { 'music_summary': { 'staging_raw_table': 'staging_sub1', 'file_name': 'sub1_filename.csv', 'report_name': 'red_label_summary_v1_1__music_summary', }, 'non_music_summary': { 'staging_raw_table': 'staging_sub2', 'file_name': 'sub2_filename.csv', 'report_name': 'red_label_summary_v1_1__non_music_summary', }, } }, ]} @patch.object(tasks.garcon_feed_status, 'get_overall_status') @pytest.mark.parametrize( 'case_name, context, expected', [ ( 'New report, no subreports', dict( date='2022-06-01', report_name='adj_claim_summary_v1_1', reload=None, ), {'date': '2022-06-01', 'feed_name': 'youtube_monthly_adj_claim_summary_v1_1', 'report_name': 'adj_claim_summary_v1_1', 's3_archive_path': 's3://dev-cucumbers/YouTubeMonthly/reports/adj_claim_summary_v1_1/2022-06-01/', # NOQA: E501 's3_drop_path': 's3://dev-feed-drop/ftp/youtube/', 'source_files_dict': { 'files': [ { 'file_name': 'YouTube_theorchardmusic_M_20220601_ADJ_claim_summary_v1-1.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'theorchardmusic'} ] }, 'staging_raw_table': 'staging_raw_youtube_monthly_adj_claim_summary_v1_1', # NOQA: E501 'file_pattern_regexp': '.+YouTube_(theorchardmusic)_M_20220601_ADJ_claim_summary_v1-1.csv.gz', # NOQA: E501 }, ), ( 'New report, csv_header_lines specified', dict( date='2022-06-01', report_name='red_label_rawdata_video_v1_1', reload=None, ), {'date': '2022-06-01', 'feed_name': 'youtube_monthly_red_label_rawdata_video_v1_1', 'report_name': 'red_label_rawdata_video_v1_1', 's3_archive_path': 's3://dev-cucumbers/YouTubeMonthly/reports/red_label_rawdata_video_v1_1/2022-06-01/', # NOQA: E501 's3_drop_path': 's3://dev-feed-drop/ftp/youtube/', 'source_files_dict': { 'files': [ { 'file_name': 'YouTube_theorchardmusic_M_20220601_20220630_red_label_rawdata_video_v1-1.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 2, 'mcn_account': 'theorchardmusic'} ] }, 'staging_raw_table': 'staging_raw_youtube_monthly_red_label_rawdata_video_v1_1__video_claim_raw_report', # NOQA: E501 'file_pattern_regexp': '.+YouTube_(theorchardmusic)_M_20220601_20220630_red_label_rawdata_video_v1-1.csv.gz', # NOQA: E501 }, ), ( 'New report WITH subreports', dict( date='2022-06-01', report_name='adjustment_red_music_rawdata_video_v1_1', reload=None, ), dict( feed_name='youtube_monthly_adjustment_red_music_rawdata_video_v1_1', # NOQA: E501 staging_raw_table=None, s3_drop_path='s3://dev-feed-drop/ftp/youtube/', s3_archive_path='s3://dev-cucumbers/' 'YouTubeMonthly/reports/adjustment_red_music_rawdata_video_v1_1/2022-06-01/', # NOQA: E501 source_files_dict=dict(files=[ { 'file_name': 'YouTube_dmgi_M_20220601_20220630_adjustment_red_music_rawdata_video_v1-1.csv.gz', # NOQA: E501 'mcn_account': 'dmgi', 'file_size': None, 'csv_header_lines': 1, 'subreports': { 'video_claim_raw_report': { 'file_name': 'YouTube_dmgi_M_20220601_20220630_adjustment_red_music_rawdata_video_v1-1__video_claim_raw_report.csv', # NOQA: E501 'report_title': 'Video Claim Raw Report', 'report_name': 'adjustment_red_music_rawdata_video_v1_1__video_claim_raw_report', # NOQA: E501 'staging_raw_table': 'staging_raw_youtube_monthly_adjustment_red_music_rawdata_video_v1_1__video_claim_raw_report' # NOQA: E501 } }, }, ]), date='2022-06-01', report_name='adjustment_red_music_rawdata_video_v1_1', file_pattern_regexp='.+YouTube_(dmgi)_M_20220601_20220630_adjustment_red_music_rawdata_video_v1-1.csv.gz', # NOQA: E501 ), ), ( 'New report with multiple accounts (and date in middle)', dict( date='2022-06-15', report_name='red_rawdata_video_v1_1', reload=None, ), dict( date='2022-06-01', feed_name='youtube_monthly_red_rawdata_video_v1_1', file_pattern_regexp='.+YouTube_(dmgi|theorchardmusic)_M_20220601_20220630_red_rawdata_video_v1-1.csv.gz', # NOQA: E501 report_name='red_rawdata_video_v1_1', s3_archive_path='s3://dev-cucumbers/YouTubeMonthly/reports/red_rawdata_video_v1_1/2022-06-01/', # NOQA: E501 s3_drop_path='s3://dev-feed-drop/ftp/youtube/', source_files_dict={ 'files': [ { 'file_name': 'YouTube_dmgi_M_20220601_20220630_red_rawdata_video_v1-1.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'dmgi', 'subreports': { 'video_claim_raw_report': { 'file_name': 'YouTube_dmgi_M_20220601_20220630_red_rawdata_video_v1-1__video_claim_raw_report.csv', # NOQA: E501 'report_name': 'red_rawdata_video_v1_1__video_claim_raw_report', # NOQA: E501 'report_title': 'Video Claim Raw Report', # NOQA: E501 'staging_raw_table': 'staging_raw_youtube_monthly_red_rawdata_video_v1_1__video_claim_raw_report' # NOQA: E501 } } }, { 'file_name': 'YouTube_theorchardmusic_M_20220601_20220630_red_rawdata_video_v1-1.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'theorchardmusic', 'subreports': { 'video_claim_raw_report': { 'file_name': 'YouTube_theorchardmusic_M_20220601_20220630_red_rawdata_video_v1-1__video_claim_raw_report.csv', # NOQA: E501 'report_name': 'red_rawdata_video_v1_1__video_claim_raw_report', # NOQA: E501 'report_title': 'Video Claim Raw Report', # NOQA: E501 'staging_raw_table': 'staging_raw_youtube_monthly_red_rawdata_video_v1_1__video_claim_raw_report' # NOQA: E501 } } } ] }, staging_raw_table=None) ), ( 'New report with multiple accounts (filtered)', dict( date='2022-06-01', report_name='red_rawdata_video_v1_1', reload=None, accounts='dmgi,theorchardmusic' ), dict( date='2022-06-01', feed_name='youtube_monthly_red_rawdata_video_v1_1', file_pattern_regexp='.+YouTube_(dmgi|theorchardmusic)_M_20220601_20220630_red_rawdata_video_v1-1.csv.gz', # NOQA: E501 report_name='red_rawdata_video_v1_1', s3_archive_path='s3://dev-cucumbers/YouTubeMonthly/reports/red_rawdata_video_v1_1/2022-06-01/', # NOQA: E501 s3_drop_path='s3://dev-feed-drop/ftp/youtube/', source_files_dict={ 'files': [ { 'file_name': 'YouTube_dmgi_M_20220601_20220630_red_rawdata_video_v1-1.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'dmgi', 'subreports': { 'video_claim_raw_report': { 'file_name': 'YouTube_dmgi_M_20220601_20220630_red_rawdata_video_v1-1__video_claim_raw_report.csv', # NOQA: E501 'report_name': 'red_rawdata_video_v1_1__video_claim_raw_report', # NOQA: E501 'report_title': 'Video Claim Raw Report', # NOQA: E501 'staging_raw_table': 'staging_raw_youtube_monthly_red_rawdata_video_v1_1__video_claim_raw_report' # NOQA: E501 } } }, { 'file_name': 'YouTube_theorchardmusic_M_20220601_20220630_red_rawdata_video_v1-1.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'theorchardmusic', 'subreports': { 'video_claim_raw_report': { 'file_name': 'YouTube_theorchardmusic_M_20220601_20220630_red_rawdata_video_v1-1__video_claim_raw_report.csv', # NOQA: E501 'report_name': 'red_rawdata_video_v1_1__video_claim_raw_report', # NOQA: E501 'report_title': 'Video Claim Raw Report', # NOQA: E501 'staging_raw_table': 'staging_raw_youtube_monthly_red_rawdata_video_v1_1__video_claim_raw_report' # NOQA: E501 } } }, ] }, staging_raw_table=None) ), ( 'Legacy report', dict( date='2022-06-01', report_name='legacy_monthly', reload=None, ), {'date': '2022-06-01', 'feed_name': 'youtube_monthly_legacy_monthly', 'report_name': 'legacy_monthly', 's3_archive_path': 's3://dev-cucumbers/YouTubeMonthly/reports/legacy_monthly/2022-06-01/', # NOQA: E501 's3_drop_path': 's3://dev-feed-drop/ftp/youtube/', 'source_files_dict': { 'files': [ { 'file_name': 'YouTube_dmgi_M_20220601_20220630_rawdata_v1-0.csv.zip', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'dmgi' }, { 'file_name': 'YouTube_theorchardmusic_M_20220601_20220630_rawdata_v1-0.csv.gz', # NOQA: E501 'file_size': None, 'csv_header_lines': 1, 'mcn_account': 'theorchardmusic'} ] }, 'staging_raw_table': 'staging_raw_youtube_monthly_v2', 'file_pattern_regexp': '.+YouTube_(dmgi|theorchardmusic)_M_20220601_20220630_rawdata_v1-0.csv'}, # NOQA: E501 ), ]) def test_bootstrap_report( get_overall_status_mock, case_name, context, expected): """Test boostrap.""" context.update({'activity': MagicMock()}) result = tasks.bootstrap(**context) assert result == expected get_overall_status_mock.assert_called_with( f"youtube_monthly_{expected['report_name']}", expected['date'] ) @patch.object(tasks.garcon_feed_status, 'get_overall_status', return_value='INGESTED') def test_bootstrap_report_already_ingested( get_overall_status_mock): """Test boostrap.""" context = dict( activity=MagicMock(), date='2022-06-01', report_name='red_label_summary_v1_1', reload=None, ) expected = { 'stop': True, 'message': 'youtube_monthly_red_label_summary_v1_1 is already ingested for ' '2022-06-01' } result = tasks.bootstrap(**context) assert result == expected get_overall_status_mock.assert_called_with( f"youtube_monthly_{context['report_name']}", context['date'] ) @pytest.mark.parametrize( 'case_name, context, expected', [ [ 'Typo in account name', dict( report_name='red_rawdata_video_v1_1', accounts='thorchardmusic' ), ValueError ], [ 'Report not process this account', dict( report_name='red_music_rawdata_video_v1_1', accounts='theorchardmusic' ), ValueError ], ]) def test_bootstrap_report_negative(case_name, context, expected): """Test boostrap.""" context.update( { 'activity': MagicMock(), 'date': '2023-01-01', 'reload': '', } ) with pytest.raises(expected): tasks.bootstrap(**context) @patch.object(tasks, 's3') @patch.object(tasks, 'TemporaryDirectory') def test_move_and_extract_files(temp_dir_mock, mock_s3): """Test if files moves from archives to temp folder and convert to gz.""" activity = MagicMock() temp_dir_mock.return_value.__enter__.return_value = '/tmp/dir_no_slash' source_files_dict = {'files': [ {'file_name': 'test_zip.zip'}, {'file_name': 'test_gz.gz'}, {'file_name': 'test_csv.csv'}, ]} s3_archive_path = 's3://TestCucumber/archive/' s3_temp_path = 's3://TestCucumber/staging/' tasks.move_and_extract_files( activity=activity, source_files_dict=source_files_dict, old_s3_path=s3_archive_path, new_s3_path=s3_temp_path, ) assert mock_s3.convert_zip_to_gzip_on_s3.call_args_list == [ call(activity, zip_s3_path='s3://TestCucumber/archive/test_zip.zip', gz_s3_path='s3://TestCucumber/staging/test_zip.gz', local_temp_dir='/tmp/dir_no_slash/') ] assert mock_s3.copy_s3_key.call_args_list == [ call(old_s3_path='s3://TestCucumber/archive/test_gz.gz', new_s3_path='s3://TestCucumber/staging/test_gz.gz'), call(old_s3_path='s3://TestCucumber/archive/test_csv.csv', new_s3_path='s3://TestCucumber/staging/test_csv.csv'), ] @patch.object(tasks, 'extract_reports_from_multireport_file') @patch.object(tasks.s3, 'download_from_s3') @patch.object(tasks.s3, 'upload_to_s3') @patch.object(tasks.s3, 'copy_s3_key') @patch.object(tasks, 'TemporaryDirectory') def test_move_and_extract_files_multireport( temp_dir_mock, mock_copy_s3_key, mock_upload_to_s3, mock_download_from_s3, mock_extract_reports_from_multireport_file, ): """Test if files moves from archives to temp folder and convert to gz.""" activity = MagicMock() temp_dir_mock.return_value.__enter__.return_value = '/tmp/dir_no_slash' mock_extract_reports_from_multireport_file.return_value = { 'files': ['1.csv', '2.csv'] } source_files_dict = {'files': [ { 'file_name': 'with_sub_reports.csv.gz', 'subreports': { 'sub_report1': { 'report_title': 'Sub Report name inside File', 'file_name': '1.csv' }, 'sub_report2': { 'report_title': 'Sub Report2 name inside File', 'file_name': '2.csv' }, }, }, ]} s3_archive_path = 's3://TestCucumber/archive/' s3_temp_path = 's3://TestTempCucumber/staging/' tasks.move_and_extract_files( activity=activity, source_files_dict=source_files_dict, old_s3_path=s3_archive_path, new_s3_path=s3_temp_path, ) assert mock_copy_s3_key.call_args_list == [ call(old_s3_path='s3://TestCucumber/archive/with_sub_reports.csv.gz', # noqa: E501 new_s3_path='s3://TestTempCucumber/staging/with_sub_reports.csv.gz') # noqa: E501 ] assert mock_download_from_s3.call_args_list == [ call('TestCucumber', 'archive/with_sub_reports.csv.gz', file_path='/tmp/dir_no_slash/with_sub_reports.csv.gz'), # NOQA: E501 ] assert mock_extract_reports_from_multireport_file.call_args_list == [ call( activity, source_filename='/tmp/dir_no_slash/with_sub_reports.csv.gz', target_directory='/tmp/dir_no_slash/', report_filename_map={ 'Sub Report name inside File': '1.csv', 'Sub Report2 name inside File': '2.csv' } ), ] assert mock_upload_to_s3.call_args_list == [ call(file_path='/tmp/dir_no_slash/1.csv', bucket_name='TestTempCucumber', object_key='staging/1.csv'), # noqa: E501 call(file_path='/tmp/dir_no_slash/2.csv', bucket_name='TestTempCucumber', object_key='staging/2.csv'), # noqa: E501 ] @pytest.mark.parametrize( 'source_filename', [ 'YouTube_theorchardmusic_M_20220801_20220831_red_label_summary_v1-1.csv', # NOQA: E501 'YouTube_theorchardmusic_M_20220801_20220831_red_label_summary_v1-1.csv.gz' # NOQA: E501 ] ) def test_extract_reports_from_multireport_file(tmp_path, source_filename): """Test extract reports.""" this_dir = Path(__file__).parent source_file = this_dir / 'files' / source_filename report_filename_map = { 'Subscriber Summary': 'subscriber_summary.csv', 'Music Summary': 'music_summary.csv', 'Non Music Summary': 'non_music_summary.csv.gz', } result = tasks.extract_reports_from_multireport_file( activity=MagicMock(), source_filename=str(source_file), target_directory=str(tmp_path), report_filename_map=report_filename_map, ) expected_result = { 'files': [ 'subscriber_summary.csv', 'music_summary.csv', 'non_music_summary.csv.gz'] } assert result == expected_result def read_content(file_path): if str(file_path).endswith('.gz'): with gzip.open(file_path, mode='rt') as file: return file.readlines() else: with open(file_path, mode='rt') as file: return file.readlines() for file in expected_result['files']: report_file = tmp_path / file expected_file = this_dir / 'files' / file report_content = read_content(report_file) expected_content = read_content(expected_file) assert report_content == expected_content @pytest.mark.parametrize( 'report_filename_map, expected', [ ({ 'Subscriber Summary': 'subscriber_summary.csv', # missing report 'Non Music Summary': 'non_music_summary.csv', }, ValueError), ({ 'Subscriber Summary': 'subscriber_summary.csv', 'Music Summary': 'music_summary.csv', # extra report 'Extra report': 'extra_report.csv', 'Non Music Summary': 'non_music_summary.csv', }, ValueError), ] ) def test_extract_reports_from_multireport_file_negative( tmp_path, report_filename_map, expected): """Test extract reports.""" this_dir = Path(__file__).parent source_file = this_dir / 'files' / 'YouTube_theorchardmusic_M_20220801_20220831_red_label_summary_v1-1.csv' # noqa: E501 with pytest.raises(expected): tasks.extract_reports_from_multireport_file( activity=MagicMock(), source_filename=str(source_file), target_directory=str(tmp_path), report_filename_map=report_filename_map, ) @patch.object(tasks, 'YoutubeMonthlySF') @patch.object(tasks, 'task_status') def test_load_staging_raw_table_reports(mock_task_status, mock_executor_class): """Test load staging raw.""" activity = MagicMock() executor_mock = mock_executor_class.return_value.__enter__.return_value mock_task_status.is_completed_task.return_value = False report_name = 'test_report' staging_raw_table_name = 'staging_raw_table' tasks.load_staging_raw_table_reports( activity, feed_name='test_feed_name', date='2022-11-01', s3_dir_path='s3://temp-bucket/location/', report_name=report_name, staging_raw_table_name=staging_raw_table_name, source_files_dict=SOURCE_FILES_DICT, ) assert executor_mock.create_temp_staging_raw_table.call_args_list == [ call(report_name='test_report', table_name='temp_staging_raw_table_2022_11_01'), call(report_name='test_report', table_name='temp_staging_raw_table_2022_11_01'), ] assert executor_mock.load_temp_staging_raw_table.call_args_list == [ call( temp_staging_raw_table='temp_staging_raw_table_2022_11_01', # noqa: E501 filenames=['source_orchard.csv.gz'], s3_dir_path='s3://temp-bucket/location/', lines_to_skip=2, ), call( temp_staging_raw_table='temp_staging_raw_table_2022_11_01', # noqa: E501 filenames=['source_dmgi.csv.gz'], s3_dir_path='s3://temp-bucket/location/', lines_to_skip=1, ), ] assert executor_mock.clean_staging_raw_table.call_args_list == [ call(staging_raw_table='staging_raw_table', download_date='2022-11-01', mcn_account='theorchardmusic'), call(staging_raw_table='staging_raw_table', download_date='2022-11-01', mcn_account='dmgi'), ] assert executor_mock.load_staging_raw_table.call_args_list == [ call(staging_raw_table='staging_raw_table', temp_staging_raw_table='temp_staging_raw_table_2022_11_01', report_name='test_report', mcn_account='theorchardmusic', date='2022-11-01', filenames=['source_orchard.csv.gz']), call(staging_raw_table='staging_raw_table', temp_staging_raw_table='temp_staging_raw_table_2022_11_01', report_name='test_report', mcn_account='dmgi', date='2022-11-01', filenames=['source_dmgi.csv.gz'])] @patch.object(tasks, 'YoutubeMonthlySF') @patch.object(tasks, 'task_status') def test_load_staging_raw_table_reports_multireport_file( mock_task_status, mock_executor_class): """Test load staging raw for multireport youtube report file.""" activity = MagicMock() executor_mock = mock_executor_class.return_value.__enter__.return_value mock_task_status.is_completed_task.return_value = False report_name = 'test_report_multi' staging_raw_table_name = '' # each subreport has own table tasks.load_staging_raw_table_reports( activity, feed_name='test_feed_name', date='2022-11-01', s3_dir_path='s3://temp-bucket/location/', report_name=report_name, staging_raw_table_name=staging_raw_table_name, source_files_dict=SOURCE_FILES_DICT_MULTIREPORT, ) assert executor_mock.create_temp_staging_raw_table.call_args_list == [ call(report_name='red_label_summary_v1_1__music_summary', table_name='temp_staging_sub1_2022_11_01'), call(report_name='red_label_summary_v1_1__non_music_summary', table_name='temp_staging_sub2_2022_11_01'), ] assert executor_mock.load_temp_staging_raw_table.call_args_list == [ call( temp_staging_raw_table='temp_staging_sub1_2022_11_01', filenames=['sub1_filename.csv'], s3_dir_path='s3://temp-bucket/location/', lines_to_skip=1, ), call( temp_staging_raw_table='temp_staging_sub2_2022_11_01', filenames=['sub2_filename.csv'], s3_dir_path='s3://temp-bucket/location/', lines_to_skip=1, ), ] assert executor_mock.clean_staging_raw_table.call_args_list == [ call(staging_raw_table='staging_sub1', download_date='2022-11-01', mcn_account='theorchardmusic'), call(staging_raw_table='staging_sub2', download_date='2022-11-01', mcn_account='theorchardmusic'), ] assert executor_mock.load_staging_raw_table.call_args_list == [ call(staging_raw_table='staging_sub1', temp_staging_raw_table='temp_staging_sub1_2022_11_01', report_name='red_label_summary_v1_1__music_summary', mcn_account='theorchardmusic', date='2022-11-01', filenames=['sub1_filename.csv']), call(staging_raw_table='staging_sub2', temp_staging_raw_table='temp_staging_sub2_2022_11_01', report_name='red_label_summary_v1_1__non_music_summary', mcn_account='theorchardmusic', date='2022-11-01', filenames=['sub2_filename.csv'])]