"""Unit tests for tasks for the Chartmetric Socials Ingestion Workflow.""" from unittest.mock import call, patch from unittest.mock import MagicMock from freezegun import freeze_time import pytest from feed_ingestion.flows.chartmetric_socials import config from feed_ingestion.flows.chartmetric_socials import tasks CHARTMETRIC_SOCIALS = 'chartmetric_socials' @freeze_time('2020-01-01') def test_bootstrap_default_date(monkeypatch): """Test that bootstrap returns the default date.""" date = None expected = { 'feed_name': CHARTMETRIC_SOCIALS, 'date': '2019-12-31', 'date_limit': '3000-01-01', 'platform_names': config.all_platform_names, 'reload': False, 'ingestion_started_at': '2020-01-01 00:00:00' } result = tasks.bootstrap(MagicMock(), date) assert result == expected @freeze_time('2020-01-01') def test_bootstrap_context_platform(monkeypatch): """Test that bootstrap returns the default date.""" date = None platform_names = 'instagram,twitter' expected = { 'feed_name': CHARTMETRIC_SOCIALS, 'date': '2019-12-31', 'date_limit': '3000-01-01', 'platform_names': ['instagram', 'twitter'], 'reload': False, 'ingestion_started_at': '2020-01-01 00:00:00' } result = tasks.bootstrap(MagicMock(), date, platform_names=platform_names) assert result == expected @freeze_time('2020-01-01') def test_bootstrap_context_date(monkeypatch): """Test that bootstrap returns the context date.""" date = '2019-01-01' expected = { 'feed_name': CHARTMETRIC_SOCIALS, 'date': '2019-01-01', 'date_limit': '3000-01-01', 'platform_names': config.all_platform_names, 'reload': False, 'ingestion_started_at': '2020-01-01 00:00:00' } result = tasks.bootstrap(MagicMock(), date) assert result == expected @freeze_time('2020-01-01') def test_bootstrap_context_date_with_reload_as_string(monkeypatch): """Test that bootstrap reload False.""" date = '2019-01-01' expected = { 'feed_name': CHARTMETRIC_SOCIALS, 'date': '2019-01-01', 'date_limit': '3000-01-01', 'platform_names': config.all_platform_names, 'reload': False, 'ingestion_started_at': '2020-01-01 00:00:00' } result = tasks.bootstrap(MagicMock(), date, reload='False') assert result == expected @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_neo4j_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.Neo4jExecutor') def test_ingest_social_accounts( mock_neo4j_executor_class, mock_get_neo4j_config, mock_sf_executor_class, mock_get_sf_config): """Test ingest spotify raw data.""" mock_get_sf_config.return_value = {'db': 'DB', 'schema': 'SCHEMA'} mock_neo4j_executor = MagicMock() mock_neo4j_executor_class.return_value.__enter__.return_value = \ mock_neo4j_executor mock_sf_executor = MagicMock() mock_sf_executor_class.return_value.__enter__.return_value = \ mock_sf_executor row_data = MagicMock(name='raw_data') mock_sf_executor.fetchall_dict_query.side_effect = ([row_data], []) mock_neo4j_config = {} mock_get_neo4j_config.return_value = {} query_name = 'ingest_accounts' neo4j_params = { 'rows': [row_data], 'platform_name': 'spotify', 'created_by': 'swf-feed-ingestion/chartmetric-socials/' 'ingest-spotify-social-accounts'} platform_name = 'spotify' sf_params = { 'table_name': 'accounts_chartmetric_socials_spotify', 'limit': 1500, 'offset': 0, } # second call args should shift offset sf_params_2 = dict(sf_params) sf_params_2['offset'] = 1500 tasks.ingest_social_accounts( MagicMock(), CHARTMETRIC_SOCIALS, platform_name) mock_get_neo4j_config.assert_called_once_with(CHARTMETRIC_SOCIALS) mock_neo4j_executor_class.assert_called_once_with( CHARTMETRIC_SOCIALS, mock_neo4j_config) mock_neo4j_executor.execute_write_query.assert_called_once_with( query_name.format(platform_name=platform_name), neo4j_params) assert mock_sf_executor.fetchall_dict_query.call_args_list == [ call('get_social_accounts', **sf_params), call('get_social_accounts', **sf_params_2), ] @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_neo4j_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.Neo4jExecutor') def test_delete_created_pending_social_accounts( mock_neo4j_executor_class, mock_get_neo4j_config): """Test delete created pending social accounts.""" mock_neo4j_executor = MagicMock() mock_neo4j_executor_class.return_value.__enter__.return_value = \ mock_neo4j_executor mock_neo4j_config = {} mock_get_neo4j_config.return_value = {} query_name = 'delete_created_pending_social_accounts' tasks.delete_pending_social_accounts( MagicMock(), CHARTMETRIC_SOCIALS, query_name) mock_get_neo4j_config.assert_called_once_with(CHARTMETRIC_SOCIALS) mock_neo4j_executor_class.assert_called_once_with( CHARTMETRIC_SOCIALS, mock_neo4j_config) mock_neo4j_executor.execute_write_query.assert_called_once_with( query_name) @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_neo4j_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.Neo4jExecutor') def test_delete_old_pending_social_accounts( mock_neo4j_executor_class, mock_get_neo4j_config): """Test delete old pending social accounts.""" mock_neo4j_executor = MagicMock() mock_neo4j_executor_class.return_value.__enter__.return_value = \ mock_neo4j_executor mock_neo4j_config = {} mock_get_neo4j_config.return_value = {} query_name = 'delete_old_pending_social_accounts' tasks.delete_pending_social_accounts( MagicMock(), CHARTMETRIC_SOCIALS, query_name) mock_get_neo4j_config.assert_called_once_with(CHARTMETRIC_SOCIALS) mock_neo4j_executor_class.assert_called_once_with( CHARTMETRIC_SOCIALS, mock_neo4j_config) mock_neo4j_executor.execute_write_query.assert_called_once_with( query_name) @pytest.mark.parametrize('query_name, table_name, sf_query_name', [ ('ingest_aggregate_socials_by_account', 'aggregate_by_account', 'get_aggregate_socials_by_account'), ('ingest_aggregate_socials_by_participant', 'aggregate_by_participant', 'get_aggregate_socials_by_participant'), ]) @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_neo4j_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.Neo4jExecutor') def test_ingest_aggregate_socials( mock_neo4j_executor_class, mock_get_neo4j_config, mock_sf_executor_class, mock_get_sf_config, table_name, sf_query_name, query_name): """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 row_data = MagicMock(name='raw_data') mock_sf_executor.fetchall_dict_query.side_effect = ([row_data], []) tasks.ingest_aggregate_social_data( MagicMock(), CHARTMETRIC_SOCIALS, query_name) mock_get_neo4j_config.assert_called_once_with(CHARTMETRIC_SOCIALS) mock_neo4j_executor_class.assert_called_once_with( CHARTMETRIC_SOCIALS, mock_neo4j_config) neo4j_params = { 'rows': [row_data] } mock_neo4j_executor.execute_write_query.assert_called_once_with( query_name, neo4j_params) sf_params = { 'table_name': table_name, 'limit': 1500, 'offset': 0, } # second call args should shift offset sf_params_2 = dict(sf_params) sf_params_2['offset'] = 1500 assert mock_sf_executor.fetchall_dict_query.call_args_list == [ call(sf_query_name, **sf_params), call(sf_query_name, **sf_params_2), ] assert mock_sf_executor.fetchall_dict_query.call_count == 2 @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') def test_create_aggregate_by_account_table( mock_executor_class, mock_get_sf_config): """Test create_aggregate_by_account_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' tasks.create_aggregate_by_account_table(MagicMock()) 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_aggregate_by_account_table', table_name='aggregate_by_account') @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') def test_create_main_aggregate_by_participant_table( mock_executor_class, mock_get_sf_config): """Test aggregate_socials_by_participant.""" 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' tasks.create_main_aggregate_by_participant_table(MagicMock()) mock_get_sf_config.assert_called_once_with(secrets_path) mock_executor_class.assert_called_once_with(mock_config) @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') def test_insert_fact_socials(mock_executor_class, mock_get_sf_config): """Test populate table for spotify.""" 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' platform_name = 'spotify' tasks.insert_fact_socials( MagicMock(), 'feed_name', '2020-01-01', '2020-02-01', '2020-01-01 00:00:00', platform_name, None) mock_get_sf_config.assert_called_once_with(secrets_path) mock_executor_class.assert_called_once_with(mock_config) calls = [ call( 'insert_fact_socials_spotify', fact_socials_table='fact_socials', ingest_date='2020-01-01', ingest_date_limit='2020-02-01', ingestion_started_at='2020-01-01 00:00:00', platform_name='spotify', reload_artist_with_no_stat='False'), ] mock_executor.execute_query.assert_has_calls(calls, any_order=False) @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') def test_create_fact_social_latest(mock_executor_class, mock_get_sf_config): """Test populate table for spotify.""" 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' tasks.create_fact_social_latest(MagicMock()) 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_fact_social_latest') @patch('feed_ingestion.flows.chartmetric_socials.tasks.get_sf_config') @patch('feed_ingestion.flows.chartmetric_socials.tasks.SnowflakeExecutor') def test_create_accounts_table(mock_executor_class, mock_get_sf_config): """Test populate table for spotify.""" platform_name = 'spotify' 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' tasks.create_accounts_table(MagicMock(), CHARTMETRIC_SOCIALS, 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_with( 'create_main_accounts_table', table_name='accounts_chartmetric_socials_spotify', temp_table_name='temp_accounts_chartmetric_socials_spotify', log_table_name='chartmetric_socials_log', platform_name='spotify', main_accounts_export_limit=500000 )