"""Test tasks for YouTube Asset Conflict Snowflake-only Ingestion Workflow.""" from unittest import TestCase from unittest.mock import MagicMock from unittest.mock import Mock from unittest.mock import patch import boto3 from garcon.activity import Activity from moto import mock_aws import pytest import requests from feed_ingestion.flows.youtube_asset_conflict import config from feed_ingestion.flows.youtube_asset_conflict import tasks from tests.flows.youtube_asset_conflict.fixtures import ows_territories class TestTasks(TestCase): """Test tasks.""" @patch('feed_ingestion.tasks.feed_status_tasks.has_newer_ingested_date', return_value=False) def test_bootstrap(self, mock_guard): """Test bootstrap task.""" activity_mock = MagicMock() input_date = '2017-01-01' expected_date = '2017-01-01' date_in_YYYYMMDD = input_date.replace('-', '') expected_drop_file_names = [ 'content_owner_asset_conflict_a3.ORCH.csv.gz' ] response = tasks.bootstrap(activity_mock, input_date, None) assert response['date'] == expected_date assert 'feed_name' in response assert 's3_feed_name' in response assert response['feed_name'] == config.feed_name assert response['s3_feed_name'] == config.s3_feed_name assert expected_date in response['s3_archive_path'] assert expected_date in response['s3_staging_raw_temp'] TestCase.assertListEqual( self, sorted(expected_drop_file_names), sorted(response['drop_file_names']) ) assert response['staging_raw_table'] ==\ 'staging_raw_youtube_asset_conflict' assert response['fact_conflict_table'] ==\ 'fact_conflict' assert response[ 'youtube_asset_conflict_by_territory_temp_table'] ==\ 'youtube_asset_conflict_by_territory_{datestamp}'.format( datestamp=date_in_YYYYMMDD) assert response['fact_conflict_temp_table'] ==\ 'temp_fact_conflict_{datestamp}'.format(datestamp=date_in_YYYYMMDD) assert response['staging_raw_youtube_asset_report_table'] ==\ 'staging_raw_youtube_asset_report' assert response['registry_table'] == 'registry' assert response['orchard_account'] == 'ORCH' assert response['stop'] is False @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_create_staging_raw_temp_table(self, mock_sf_executor): """Test create_staging_raw_temp_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'staging_raw_temp_table': config.snowflake_table_names.get( 'staging_raw_temp').format( account='account', datestamp='20170101') } sf_executor = MagicMock() sf_executor.create_staging_raw_temp_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.create_temp_staging_table, context, task_name='create_temp_staging_table_staging_raw_' 'youtube_asset_conflict_account_20170101') self._run_task( tasks.create_temp_staging_table, context, task_name='create_temp_staging_table_staging_raw_' 'youtube_asset_conflict_account_20170101') sf_executor.create_staging_raw_temp_table.assert_called_with( context.get('staging_raw_temp_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_load_temp_staging_table(self, mock_sf_executor): """Test load_temp_staging_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'staging_raw_temp_table': config.snowflake_table_names.get( 'staging_raw_temp').format( account='account', datestamp='20170101'), 'key_dir': 'path/', 'drop_file': 'filename' } sf_executor = MagicMock() sf_executor.create_staging_raw_temp_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.load_temp_staging_table, context, task_name='load_temp_staging_table_staging_raw_' 'youtube_asset_conflict_account_20170101', ) self._run_task( tasks.load_temp_staging_table, context, task_name='load_temp_staging_table_staging_raw_' 'youtube_asset_conflict_account_20170101', ) sf_executor.load_staging_raw_temp_table.assert_called_with( context.get('staging_raw_temp_table'), context.get('key_dir') + context.get('drop_file')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_truncate_staging_raw_table(self, mock_sf_executor): """Test truncate_staging_raw_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'staging_raw_table': config.snowflake_table_names.get( 'staging_raw') } sf_executor = MagicMock() sf_executor.truncate_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.truncate_staging_raw_table, context) self._run_task( tasks.truncate_staging_raw_table, context) sf_executor.truncate_table.assert_called_with( context.get('staging_raw_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.helpers') @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_insert_into_staging_raw_table( self, mock_sf_executor, mock_helpers): """Test insert_into_staging_raw_table task.""" file_name = 'file_name' content_owner = 'some_owner' some_bucket = 'bucket' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'staging_raw_table': config.snowflake_table_names.get( 'staging_raw'), 'staging_raw_temp_table': config.snowflake_table_names.get( 'staging_raw_temp').format( account='account', datestamp='20170101'), 'drop_file': file_name, 's3_staging_raw_temp': 's3://{}/'.format(some_bucket), 'content_owner': content_owner } file_size = '2' sf_executor = MagicMock() sf_executor.insert_into_staging_raw_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor mock_helpers.get_filesize = MagicMock(return_value=file_size) self._run_completed_task( tasks.insert_into_staging_raw_table, context, task_name='insert_into_staging_raw_table_from_staging_raw_' 'youtube_asset_conflict_account_20170101' ) self._run_task( tasks.insert_into_staging_raw_table, context, task_name='insert_into_staging_raw_table_from_staging_raw_' 'youtube_asset_conflict_account_20170101' ) sf_executor.insert_into_staging_raw_table.assert_called_with( context.get('staging_raw_temp_table'), context.get('staging_raw_table'), file_name, file_size, context.get('date'), content_owner) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_create_fact_conflict_table(self, mock_sf_executor): """Test create_fact_conflict_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp='20170101'), } sf_executor = MagicMock() sf_executor.create_fact_conflict_temp_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.create_fact_conflict_temp_table, context, task_name='create_temp_fact_conflict_20170101') self._run_task( tasks.create_fact_conflict_temp_table, context, task_name='create_temp_fact_conflict_20170101') sf_executor.create_fact_conflict_temp_table.assert_called_with( context.get('fact_conflict_temp_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_insert_overwrite_into_fact_conflict_temp_table( self, mock_sf_executor): """Test insert_overwrite_into_fact_conflict_temp.""" datestamp = '20170101' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'art_relations_db': config.art_relations_db, 'art_relations_schema': config.art_relations_schema, 'asset_report_schema': config.asset_report_schema, 'registry_schema': config.registry_schema, 'registry_table': config.snowflake_table_names.get( 'registry'), 'staging_raw_youtube_asset_report_table': config.snowflake_table_names.get( 'staging_raw_youtube_asset_report'), 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp), 'yact_table': config.snowflake_table_names.get( 'youtube_asset_conflict_by_territory_temp'), 'track_table': config.snowflake_table_names.get('track'), 'releases_table': config.snowflake_table_names.get('releases'), 'artist_info_table': config.snowflake_table_names.get('artist_info'), 'territory_standard': config.territory_standard, 'asset_type': config.sound_recording_asset_type, 'time_zone': config.time_zone, 'orchard_account': config.accounts['ORCHARD']['file_label'] } sf_executor = MagicMock() sf_executor.insert_overwrite_into_fact_conflict_temp_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.insert_overwrite_into_fact_conflict_temp_table, context, task_name='insert overwrite into temp_fact_conflict_20170101' ) self._run_task( tasks.insert_overwrite_into_fact_conflict_temp_table, context, task_name='insert overwrite into temp_fact_conflict_20170101' ) sf_executor.insert_overwrite_into_fact_conflict_temp_table\ .assert_called_with( context.get('art_relations_db'), context.get('art_relations_schema'), context.get('asset_report_schema'), context.get('registry_schema'), context.get('fact_conflict_temp_table'), context.get( 'yact_table'), context.get('staging_raw_youtube_asset_report_table'), context.get('registry_table'), context.get('track_table'), context.get('releases_table'), context.get('artist_info_table'), context.get('territory_standard'), context.get('asset_type'), context.get('time_zone'), context.get('orchard_account')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_drop_create_territories_temp_table(self, mock_sf_executor): """Test drop_create_territories_temp_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01' } territories_table = config.snowflake_table_names.get( 'territories_temp') sf_executor = MagicMock() sf_executor.truncate_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.drop_create_territories_temp_table, context) self._run_task( tasks.drop_create_territories_temp_table, context) sf_executor.drop_table.assert_called_with(territories_table) sf_executor.create_territories_temp_table.assert_called_with( territories_table) @patch('feed_ingestion.flows.youtube_asset_conflict.models.' 'ows_territories.request.process') @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_fill_territories_temp_table( self, mock_sf_executor, mock_ows_territories): """Test fill_territories_temp_table task.""" standard = config.territory_standard context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'standard': standard } territories_table = config.snowflake_table_names.get( 'territories_temp') sf_executor = MagicMock() sf_executor.fill_territories_temp_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor mock_ows_territories.return_value = MagicMock( spec=requests.models.Response, status_code=200, json=lambda: {'items': ows_territories.TEST_TERRITORIES}) self._run_completed_task( tasks.fill_territories_temp_table, context) self._run_task( tasks.fill_territories_temp_table, context) sf_executor.fill_territories_temp_table.assert_called_with( territories_table, ows_territories.TEST_TERRITORIES) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_create_staging_raw_yact_table_task( self, mock_sf_executor): """Test create_youtube_asset_conflict_by_territory_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'yact_table': config.snowflake_table_names.get( 'youtube_asset_conflict_by_territory_temp').format( datestamp='20170101') } sf_executor = MagicMock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.create_youtube_asset_conflict_by_territory_table, context) self._run_task( tasks.create_youtube_asset_conflict_by_territory_table, context) (sf_executor. create_youtube_asset_conflict_by_territory_table. assert_called_with(context.get('yact_table'))) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_fill_youtube_asset_conflict_by_territory_table_task( self, mock_sf_executor): """Test fill_youtube_asset_conflict_by_territory_table task.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'yact_table': config.snowflake_table_names.get( 'youtube_asset_conflict_by_territory_temp').format( datestamp='20170101'), 'staging_raw_table': config.snowflake_table_names.get( 'staging_raw') } territories_table = config.snowflake_table_names.get( 'territories_temp') sf_executor = MagicMock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.fill_youtube_asset_conflict_by_territory_table, context) self._run_task( tasks.fill_youtube_asset_conflict_by_territory_table, context) (sf_executor. fill_youtube_asset_conflict_by_territory_table.assert_called_with( context.get('yact_table'), context.get('staging_raw_table'), territories_table)) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_check_size_of_fact_conflict_temp_table_exception( self, mock_sf_executor): """Test check_size_of_fact_conflict_temp_table is under the minimum.""" with pytest.raises(Exception) as excinfo: datestamp = '20170101' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp) } sf_executor = MagicMock() sf_executor.check_size_of_fact_conflict_temp_table = Mock( return_value=config.fact_conflict_temp_size_exception_threshold ) sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.check_size_of_fact_conflict_temp_table, context) self._run_task( tasks.check_size_of_fact_conflict_temp_table, context) assert ('Not enough data in the fact conflict temp table, exiting' in str(excinfo.value)) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_check_size_of_fact_conflict_temp_table_no_exception( self, mock_sf_executor): """Test check_size_of_fact_conflict_temp_table is over the minimum.""" datestamp = '20170101' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp) } sf_executor = MagicMock() # with a table size above the threshold the SWF should run successfully table_size = config.fact_conflict_temp_size_exception_threshold + 1 sf_executor.check_size_of_fact_conflict_temp_table = Mock( return_value=table_size) sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.check_size_of_fact_conflict_temp_table, context) self._run_task( tasks.check_size_of_fact_conflict_temp_table, context) sf_executor.check_size_of_fact_conflict_temp_table\ .assert_called_with( context.get('fact_conflict_temp_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_check_number_of_unresolved_conflicts_exception( self, mock_sf_executor): """Test check_number_of_unresolved_conflicts is under the minimum.""" with pytest.raises(Exception) as excinfo: context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict') } sf_executor = MagicMock() sf_executor.check_number_of_unresolved_conflicts = Mock( return_value=config.unresolved_conflicts_exception_threshold ) sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.check_number_of_unresolved_conflicts, context) self._run_task( tasks.check_number_of_unresolved_conflicts, context) assert ('Too few unresolved conflicts in fact conflict table, exiting' in str(excinfo.value)) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_check_number_of_unresolved_conflicts_no_exception( self, mock_sf_executor): """Test check_number_of_unresolved_conflicts is over the minimum.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict') } sf_executor = MagicMock() # with a number of unresolved conflicts above the threshold the SWF # should run successfully num_unresolved = config.unresolved_conflicts_exception_threshold + 1 sf_executor.check_number_of_unresolved_conflicts = Mock( return_value=num_unresolved) sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.check_number_of_unresolved_conflicts, context) self._run_task( tasks.check_number_of_unresolved_conflicts, context) sf_executor.check_number_of_unresolved_conflicts\ .assert_called_with( context.get('fact_conflict_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_insert_into_fact_conflict_table( self, mock_sf_executor): """Test insert_into_fact_conflict_table.""" datestamp = '20170101' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp), 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict' ) } sf_executor = MagicMock() sf_executor.insert_into_fact_conflict_table = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.insert_into_fact_conflict_table, context) self._run_task( tasks.insert_into_fact_conflict_table, context) sf_executor.insert_into_fact_conflict_table\ .assert_called_with( context.get('fact_conflict_table'), context.get('fact_conflict_temp_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_update_fact_conflict_yt_recent_daily_average( self, mock_sf_executor): """Test update_fact_conflict_yt_recent_daily_average.""" datestamp = '20170101' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp), 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict' ) } sf_executor = MagicMock() sf_executor.update_fact_conflict_yt_recent_daily_average = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.update_fact_conflict_yt_recent_daily_average, context) self._run_task( tasks.update_fact_conflict_yt_recent_daily_average, context) sf_executor.update_fact_conflict_yt_recent_daily_average\ .assert_called_with( context.get('fact_conflict_table'), context.get('fact_conflict_temp_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_update_fact_conflict_views_in_conflict( self, mock_sf_executor): """Test update_fact_conflict_views_in_conflict.""" datestamp = '20211230' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2021-12-30', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp), 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict' ) } sf_executor = MagicMock() sf_executor.update_fact_conflict_views_in_conflict = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.update_fact_conflict_views_in_conflict, context) self._run_task( tasks.update_fact_conflict_views_in_conflict, context) sf_executor.update_fact_conflict_views_in_conflict \ .assert_called_with( context.get('fact_conflict_table'), context.get('fact_conflict_temp_table')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_update_fact_conflict_resolved_datetime( self, mock_sf_executor): """Test update_fact_conflict_resolved_datetime.""" datestamp = '20170101' context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=datestamp), 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict' ), 'time_zone': config.time_zone } sf_executor = MagicMock() sf_executor.update_fact_conflict_resolved_datetime = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.update_fact_conflict_resolved_datetime, context) self._run_task( tasks.update_fact_conflict_resolved_datetime, context) sf_executor.update_fact_conflict_resolved_datetime\ .assert_called_with( context.get('fact_conflict_table'), context.get('fact_conflict_temp_table'), context.get('time_zone')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') def test_reset_es_indexed_for_partially_resolved_conflicts( self, mock_sf_executor): """Test reset_es_indexed_for_partially_resolved_conflicts.""" context = { 'activity': Activity(boto3.client('swf', 'us-east-1')), 'date': '2017-01-01', 'fact_conflict_table': config.snowflake_table_names.get( 'fact_conflict' ), 'time_zone': config.time_zone } sf_executor = MagicMock() sf_executor.update_fact_conflict_resolved_datetime = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor self._run_completed_task( tasks.reset_es_indexed_for_partially_resolved_conflicts, context) self._run_task( tasks.reset_es_indexed_for_partially_resolved_conflicts, context) sf_executor.reset_es_indexed_for_partially_resolved_conflicts \ .assert_called_with( context.get('fact_conflict_table'), context.get('time_zone')) @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.task_status') def _run_completed_task( self, task, context, mock_task_status, task_name=None): """Check that task runs correctly when status is marked as completed. Args: task (callable): Garcon task to call. context (dict): Context to pass to the task. mock_task_status (MagicMock): Mock object for task status. Returns: return_value (dict or None): Return value of task. """ mock_task_status.is_completed_task.return_value = True task_name = task_name or task.__name__ return_value = task(**context) mock_task_status.is_completed_task.assert_called_with( config.feed_name, context['date'], task_name) assert not mock_task_status.mark_completed_task.called return return_value @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.task_status') @mock_aws def _run_task(self, task, context, mock_task_status, task_name=None): """Check that task runs correctly when status is not completed. Args: task (callable): Garcon task to call. context (dict): Context to pass to the task. mock_task_status (MagicMock): Mock object for task status. Returns: return_value (dict or None): Return value of task. """ mock_task_status.is_completed_task.return_value = False task_name = task_name or task.__name__ return_value = task(**context) mock_task_status.is_completed_task.assert_called_with( config.feed_name, context['date'], task_name) mock_task_status.mark_completed_task.assert_called_with( config.feed_name, context['date'], task_name) return return_value @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.' 'YouTubeAssetConflictSFExecutor') @patch('feed_ingestion.flows.youtube_asset_conflict.tasks.task_status') def test_drop_temp_table(self, mock_task_status, mock_sf_executor): """Test drop_temp_table task.""" sf_executor = MagicMock() sf_executor.update_fact_conflict_resolved_datetime = Mock() sf_executor.__enter__.return_value = sf_executor mock_sf_executor.return_value = sf_executor tasks.drop_temp_table(MagicMock(), '2017-01-01', 'test_table') (sf_executor.drop_table.assert_called_with( 'test_table'))