"""Unit tests for tasks of YouTube Bulk Reports Workflow.""" from datetime import date import re from unittest.mock import MagicMock from unittest.mock import patch from garcon_contrib.dynamo_feed_status import garcon_feed_status import pytest from feed_ingestion.flows.youtube_bulk_reports import config from feed_ingestion.flows.youtube_bulk_reports import tasks TASKS_MODULE = 'feed_ingestion.flows.youtube_bulk_reports.tasks' class TestBootstrap(object): """Test bootstrap.""" @pytest.fixture def mock_activity(self): """Return mock activity.""" return MagicMock() @pytest.fixture def context(self): """Context for YouTube Bulk Reports tasks.""" return { 'activity': MagicMock(), 'date': '2017-11-01', 'licensor': 'theorchard', 'report_name': 'cards', 'reload': False} @pytest.fixture() def group_context(self): """Context for YouTube Bulk Reports Group ingestion tasks.""" return { 'activity': MagicMock(), 'licensor': 'theorchard', 'date': '2017-11-01', 'feed_name': config.feed_name, 'report_list': ['ad_rates'], 'reload': False} @pytest.fixture def mock_executor_context(self): """Yield executor context.""" sf_executor_class_path = ( f'{TASKS_MODULE}.' 'YouTubeBulkReportsSF') with patch(sf_executor_class_path) as sf_executor: mock_executor_context = ( sf_executor.return_value.__enter__.return_value) yield mock_executor_context @pytest.fixture def expected_bootstrap_response(self, context): """Response for bootstrap task.""" return { 'date': '2017-11-01', 'archive_path': 'YouTubeBulkReports/archives/2017-11-01/theorchard/', 'credentials_path': config.credentials_paths.get('theorchard'), 'feed_name': 'youtube_bulk_reports_theorchard', 'report_name': 'cards', 'licensor': 'theorchard', 'only_download': None, 'selected_owner': None, 's3_bucket': 'dev-cucumbers', 's3_dir_path': 's3://dev-cucumbers/YouTubeBulkReports' '/archives/2017-11-01/theorchard/', 'staging_raw_table': 'cards_intermediate', 'report_status_name': 'youtube_bulk_reports_theorchard_cards', 'youtube_report_id': 'content_owner_cards_a1' } @pytest.fixture def expected_group_bootstrap_response(self, context): """Response for bootstrap group ingestion task.""" return { 'date': '2017-11-01', 'feed_name': 'youtube_bulk_reports_theorchard', 'licensor': 'theorchard', 'credentials_path': config.credentials_paths.get('theorchard'), 'cms_dict': config.orchard_content_owners_map, 'reports': [{ 'archive_path': 'YouTubeBulkReports/archives/2017-11-01/theorchard/', 'report_name': 'ad_rates', 'staging_raw_table': 'adrates_intermediate', 'report_status_name': 'youtube_bulk_reports_theorchard_ad_rates', 's3_bucket': 'dev-cucumbers', 's3_dir_path': 's3://dev-cucumbers/YouTubeBulkReports' '/archives/2017-11-01/theorchard/', 'youtube_report_id': 'content_owner_ad_rates_a1' }], } @pytest.fixture def mock_set_overall_status(self): """Yield overall status.""" overall_status_path = ( f'{TASKS_MODULE}.' 'garcon_feed_status.set_overall_status') with patch(overall_status_path) as overall_status: yield overall_status @pytest.fixture def mock_task_status(self): """Yield task status.""" task_status_path = ( f'{TASKS_MODULE}.task_status') with patch(task_status_path) as task_status: task_status.is_completed_task.return_value = False task_status.mark_completed_task = MagicMock() yield task_status @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) def test_bootstrap( self, mock_get_overall_status, context, expected_bootstrap_response): """Test bootstrap task.""" result = tasks.bootstrap(**context) assert result == expected_bootstrap_response @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) @pytest.mark.parametrize('licensor, only_download', [ ('sme', 'True'), ('theorchard', None)]) def test_bootstrap_only_download( self, mock_get_overall_status, context, licensor, only_download): """Test bootstrap task.""" context['report_name'] = 'playlist_basic' context['licensor'] = licensor result = tasks.bootstrap(**context) assert result['only_download'] == only_download @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) def test_bootstrap_report_group( self, mock_get_overall_status, group_context, expected_group_bootstrap_response): """Test bootstrap_report_group task.""" result = tasks.bootstrap_report_group(**group_context) assert result == expected_group_bootstrap_response @patch(f'{TASKS_MODULE}.date_module') @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) def test_bootstrap_no_date_is_passed( self, mock_get_overall_status, mock_date_module, context): """Date should be set to today if it is not provided.""" context['date'] = None mock_date_module.today.return_value = date(2000, 1, 1) assert tasks.bootstrap(**context)['date'] == '2000-01-01' @patch(f'{TASKS_MODULE}.date_module') @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) def test_bootstrap_report_group_no_date_is_passed( self, mock_get_overall_status, mock_date_module, group_context): """Date should be set to today if it is not provided.""" group_context['date'] = None mock_date_module.today.return_value = date(2000, 1, 1) assert tasks.bootstrap_report_group( **group_context)['date'] == '2000-01-01' @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) def test_bootstrap_date_is_passed( self, mock_get_overall_status, context, expected_bootstrap_response): """Date should be returned if it is provided.""" context['date'] = '2017-11-01' expected_bootstrap_response['date'] = '2017-11-01' assert tasks.bootstrap(**context) == expected_bootstrap_response @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_NOT_INGESTED) def test_group_bootstrap_date_is_passed( self, mock_get_overall_status, group_context, expected_group_bootstrap_response): """Date should be returned if it is provided.""" group_context['date'] = '2017-11-01' expected_group_bootstrap_response['date'] = '2017-11-01' assert tasks.bootstrap_report_group( **group_context) == expected_group_bootstrap_response @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_INGESTED) def test_bootstrap_should_return_stop_if_already_ingested( self, mock_get_overall_status, context): """Return stop when overall feed status is already ingested.""" expected_bootstrap_response = {'stop': True} result = tasks.bootstrap(**context) mock_get_overall_status.assert_called_with( '_'.join([config.feed_name, 'theorchard', 'cards']), context['date']) assert result == expected_bootstrap_response @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_DOWNLOADED) def test_bootstrap_only_downloads_should_return_stop_if_already_downloaded( self, mock_get_overall_status, context): """Return stop when overall feed status is already ingested.""" context['report_name'] = 'playlist_basic' context['licensor'] = 'sme' result = tasks.bootstrap(**context) mock_get_overall_status.assert_called_with( '_'.join([config.feed_name, 'sme', 'playlist_basic']), context['date']) expected_bootstrap_response = {'stop': True} assert result == expected_bootstrap_response @patch(f'{TASKS_MODULE}.garcon_feed_status.get_overall_status', return_value=garcon_feed_status.STATUS_INGESTED) def test_group_bootstrap_should_return_stop_if_already_ingested( self, mock_get_overall_status, group_context): """Return stop when overall feed status is already ingested.""" expected_bootstrap_response = {'stop': True} result = tasks.bootstrap_report_group(**group_context) mock_get_overall_status.assert_called_with( '_'.join([config.feed_name, 'theorchard', 'ad_rates']), group_context['date']) assert result == expected_bootstrap_response def test_sme_source_file_pattern(self): """Test generated parquet filenames created with Athena CTAS query.""" # sample filenames created by Athena Query filenames = [ '20201022_172646_00050_y255y_038fffee-d7a3-4197-b75b-a6487479db64', '20201022_172646_00050_y255y_05506ca6-f9fd-4968-b321-8fbfc298c636', '20201022_172646_00050_y255y_1d980cac-287b-424c-b2cf-06abd3f84aae', '20201022_172646_00050_y255y_24840d90-f760-4f5e-838b-f7f59c60245c', '20201022_172646_00050_y255y_37588958-24e8-4692-88f7-16cc812c5760', '20201022_172646_00050_y255y_38783bac-caa6-47dc-8c63-bb464465f428', '20201022_172646_00050_y255y_38fac22d-5519-4a76-886c-1cb51c9cbe87', '20201022_172646_00050_y255y_427b53f7-3537-43dd-a3ee-e09ac66d04bf', '20201022_172646_00050_y255y_4b6c601d-d642-4e6e-8851-2a2a118dcf23', '20201022_172646_00050_y255y_6bd57cdc-2518-47b7-8712-1ff722353261', ] file_pattern = config.sme_source_file_pattern for filename in filenames: assert re.match(file_pattern, filename), \ f"File {filename} doesn't match to sme_source_file_pattern"