"""Unit tests for tasks for the Chartmetric Socials Ingestion Workflow.""" from unittest.mock import call, MagicMock from unittest.mock import patch from freezegun import freeze_time import pytest from feed_ingestion.flows.chartmetric_socials_backfill import config from feed_ingestion.flows.chartmetric_socials_backfill import tasks FEED_NAME = 'chartmetric_socials_backfill' @freeze_time('2020-06-04') def test_bootstrap_default_date(monkeypatch): """Test that bootstrap returns the default date.""" date = None date_from = None expected = { 'feed_name': FEED_NAME, 'date': '2020-06-03', 'date_from': '2020-06-02', 'platform_names': config.all_platform_names, 'reload': False, 'ingestion_started_at': '2020-06-04 00:00:00' } result = tasks.bootstrap(MagicMock(), date, date_from) assert result == expected @freeze_time('2020-06-04') def test_bootstrap_default_date_reload(monkeypatch): """Test that bootstrap returns the default date.""" date = None date_from = None expected = { 'feed_name': FEED_NAME, 'date': '2020-06-03', 'date_from': '2020-06-02', 'platform_names': config.all_platform_names, 'reload': True, 'ingestion_started_at': '2020-06-04 00:00:00' } result = tasks.bootstrap(MagicMock(), date, date_from, None, 'True') assert result == expected @freeze_time('2020-06-04') def test_bootstrap_default_date_with_custom_platforms(monkeypatch): """Test that bootstrap returns the default date.""" date = None date_from = None platform_names = 'instagram,twitter' expected = { 'feed_name': FEED_NAME, 'date': '2020-06-03', 'date_from': '2020-06-02', 'platform_names': ['instagram', 'twitter'], 'reload': False, 'ingestion_started_at': '2020-06-04 00:00:00' } result = tasks.bootstrap(MagicMock(), date, date_from, platform_names) assert result == expected @freeze_time('2020-06-04') def test_bootstrap_context_date(monkeypatch): """Test that bootstrap returns the context date.""" date = '2020-03-01' date_from = None expected = { 'feed_name': FEED_NAME, 'date': '2020-03-01', 'date_from': '2020-02-29', 'platform_names': config.all_platform_names, 'reload': False, 'ingestion_started_at': '2020-06-04 00:00:00' } result = tasks.bootstrap(MagicMock(), date, date_from) assert result == expected @freeze_time('2020-06-04') def test_bootstrap_context_date_from(monkeypatch): """Test that bootstrap returns the context date.""" date = '2020-03-01' date_from = '2020-02-01' expected = { 'feed_name': FEED_NAME, 'date': '2020-03-01', 'date_from': '2020-02-01', 'platform_names': config.all_platform_names, 'reload': False, 'ingestion_started_at': '2020-06-04 00:00:00' } result = tasks.bootstrap(MagicMock(), date, date_from) assert result == expected @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.tasks.get_sf_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.SnowflakeExecutor') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.tasks.get_neo4j_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.tasks.Neo4jExecutor') @pytest.mark.parametrize( 'platform_name, platform_id', [ ['facebook', 0], ['instagram', 2], ['soundcloud', 7], ['tiktok', 19], ['twitter', 1], ['youtube', 3], ] ) def test_delete_removed_accounts( mock_neo4j_executor_class, mock_get_neo4j_config, mock_sf_executor_class, mock_get_sf_config, platform_name, platform_id): """Test ingest raw data.""" mock_get_sf_config.return_value = {'db': 'DB', 'schema': 'SCHEMA'} mock_sf_executor = MagicMock() mock_sf_executor_class.return_value.__enter__.return_value = \ mock_sf_executor invocation_1 = [ {'ACCOUNT_URL': 'http://1'}, {'ACCOUNT_URL': 'https://2'}, ] invocation_2 = [] mock_sf_executor.fetchmany_dict_query.side_effect = \ iter([invocation_1]), iter([invocation_2]) mock_sf_executor.execute_query.return_value = True mock_neo4j_config = {} mock_get_neo4j_config.return_value = mock_neo4j_config mock_neo4j_executor = MagicMock() mock_neo4j_executor_class.return_value.__enter__.return_value = \ mock_neo4j_executor tasks.delete_removed_accounts(MagicMock(), FEED_NAME, platform_name) mock_get_neo4j_config.assert_called_once_with(FEED_NAME) mock_neo4j_executor_class.assert_called_once_with( FEED_NAME, mock_neo4j_config) neo4j_query_name = 'delete_incorrect_accounts' neo4j_params_expected = { 'deleted_accounts': ['http://1', 'https://2'], 'platform_name': platform_name, } mock_neo4j_executor.execute_write_query.assert_called_once_with( neo4j_query_name, neo4j_params_expected) sf_params_expected = { 'platform_ID': platform_id, 'platform_name': platform_name, } assert mock_sf_executor.fetchmany_dict_query.call_args_list == [ call( query_name='get_deleted_accounts', size=2000, **sf_params_expected ), ] sf_params_expected2 = { 'deleted_accounts': ['http://1', 'https://2'], 'platform_name': platform_name, } assert mock_sf_executor.execute_query.call_args_list == [ call('delete_incorrect_accounts_fact_socials', **sf_params_expected2), ] @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.tasks.get_sf_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.SnowflakeExecutor') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.tasks.get_neo4j_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.tasks.Neo4jExecutor') def test_ingest_aggregate_social_data( mock_neo4j_executor_class, mock_get_neo4j_config, mock_sf_executor_class, mock_get_sf_config): """Test aggregation ingest queries.""" mock_neo4j_config = {} mock_get_neo4j_config.return_value = mock_neo4j_config mock_neo4j_executor = MagicMock() mock_neo4j_executor_class.return_value.__enter__.return_value = \ mock_neo4j_executor mock_get_sf_config.return_value = {'db': 'DB', 'schema': 'SCHEMA'} mock_sf_executor = MagicMock() mock_sf_executor_class.return_value.__enter__.return_value = \ mock_sf_executor first_batch = [{'key': 'value'}] second_batch = [] mock_sf_executor.fetchall_dict_query.side_effect = \ (first_batch, second_batch) date = '2020-03-01' query_name = 'ingest_aggregate_socials_by_account' tasks.ingest_aggregate_social_data(MagicMock(), FEED_NAME, query_name, date) mock_get_neo4j_config.assert_called_once_with(FEED_NAME) mock_neo4j_executor_class.assert_called_once_with( FEED_NAME, mock_neo4j_config) neo4j_params_expected = { 'rows': first_batch } mock_neo4j_executor.execute_write_query.assert_called_once_with( query_name, neo4j_params_expected) sf_params_expected = { 'limit': 2000, 'offset': 0, 'ingest_date': '2020-01-01', 'ingest_date_limit': date, 'table_name': 'aggregate_socials_backfill', } params_expected2 = { 'limit': 2000, 'offset': 2000, 'ingest_date': '2020-01-01', 'ingest_date_limit': date, 'table_name': 'aggregate_socials_backfill', } sf_query_name = 'get_for_' + query_name assert mock_sf_executor.fetchall_dict_query.call_args_list == [ call(sf_query_name, **sf_params_expected), call(sf_query_name, **params_expected2), ] @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.get_sf_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.SnowflakeExecutor') def test_create_table(mock_executor_class, mock_get_sf_config): """Test create table.""" mock_config = {} mock_get_sf_config.return_value = mock_config mock_executor = MagicMock() mock_executor_class.return_value.__enter__.return_value = mock_executor secrets_path = 'chartmetric_socials_backfill' tasks.create_table( MagicMock(), secrets_path) mock_get_sf_config.assert_called_once_with(secrets_path) mock_executor_class.assert_called_once_with(mock_config) mock_executor.execute_query.assert_called_once_with( 'create_snowflake_table', table_name='aggregate_socials_backfill') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.get_sf_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.SnowflakeExecutor') def test_populate_table(mock_executor_class, mock_get_sf_config): """Test populate table for instagram.""" mock_config = {} mock_get_sf_config.return_value = mock_config mock_executor = MagicMock() mock_executor_class.return_value.__enter__.return_value = mock_executor secrets_path = 'chartmetric_socials_backfill' platform_name = 'instagram' target_ids_table = 'tmp_target_ids_table_name' tasks.populate_table( MagicMock(), secrets_path, target_ids_table, platform_name) mock_get_sf_config.assert_called_once_with(secrets_path) mock_executor_class.assert_called_once_with(mock_config) mock_executor.execute_query.assert_called_once_with( 'populate_instagram_socials', table_name='aggregate_socials_backfill', target_ids_table='tmp_target_ids_table_name') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.get_sf_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.SnowflakeExecutor') @freeze_time('2020-06-04') def test_save_new_accounts(mock_executor_class, mock_get_sf_config): """Test save_new_accounts.""" mock_config = {} mock_get_sf_config.return_value = mock_config mock_executor = MagicMock() mock_executor_class.return_value.__enter__.return_value = mock_executor secrets_path = 'chartmetric_socials_backfill' target_ids_table = 'tmp_target_ids_table_name' ingestion_started_at = '2020-06-04 00:00:00' tasks.save_new_accounts( MagicMock(), target_ids_table, ingestion_started_at) mock_get_sf_config.assert_called_once_with(secrets_path) mock_executor_class.assert_called_once_with(mock_config) mock_executor.execute_query.assert_called_once_with( 'save_modified_accounts', cm_url_backfill_table='CM_URL_BACKFILL', target_ids_table='tmp_target_ids_table_name', ingestion_started_at=ingestion_started_at) @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.get_sf_config') @patch( 'feed_ingestion.flows.chartmetric_socials_backfill.' 'tasks.SnowflakeExecutor') def test_update_fact_socials(mock_executor_class, mock_get_sf_config): """Test populate table for instagram.""" mock_config = {} mock_get_sf_config.return_value = mock_config mock_executor = MagicMock() mock_executor_class.return_value.__enter__.return_value = mock_executor secrets_path = 'chartmetric_socials_backfill' platform_name = 'instagram' target_ids_table = 'tmp_target_ids_table_name' tasks.update_fact_socials( MagicMock(), 'feed', '2021-02-01', target_ids_table, platform_name) mock_get_sf_config.assert_called_once_with(secrets_path) mock_executor_class.assert_called_once_with(mock_config) assert mock_executor.execute_query.call_args_list == [ call('insert_fact_socials_instagram', fact_socials_table='fact_socials', platform_ID=2, platform_name='instagram', table_name='tmp_target_ids_table_name', since_date='2020-01-01', ingest_date_limit='2021-02-01') ]