"""Unit tests for Spotify Snowflake SQL executor.""" from itertools import product from unittest.mock import MagicMock from unittest.mock import patch import pytest from snowflake import connector from feed_ingestion.flows.spotify import config from feed_ingestion.flows.spotify.snowflake_executor import Spotify, \ SpotifyTheOrchardFA from tests.conftest import SubstringMatcher @pytest.fixture def mock_sql_loader(): """Return sql_loader mock.""" sql_loader_path = ( 'feed_ingestion.flows.spotify.snowflake_executor.sql_loader') with patch(sql_loader_path) as sql_loader: yield sql_loader @pytest.fixture def mock_executor(sf_config_mock, monkeypatch): """Yield executor context.""" connect_mock = MagicMock() monkeypatch.setattr(connector, 'connect', connect_mock) executor = Spotify(sf_config_mock) with patch.object(executor, 'execute', wraps=executor.execute) as \ executor.ex_mock: executor.fetchall = MagicMock() yield executor @pytest.fixture def mock_executor_theorchard(sf_config_mock, monkeypatch): """Yield executor context.""" connect_mock = MagicMock() monkeypatch.setattr(connector, 'connect', connect_mock) executor = SpotifyTheOrchardFA(sf_config_mock) with patch.object(executor, 'execute', wraps=executor.execute) as \ executor.ex_mock: executor.fetchall = MagicMock() yield executor def test_create_temp_staging_raw_table(mock_executor): """Test create_temp_staging_raw_table method.""" mock_executor.create_temp_staging_raw_table( temp_staging_raw_table='temp_table') assert mock_executor.ex_mock.call_count == 1 mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'CREATE OR REPLACE TRANSIENT', 'test_db.test_schema.temp_table']), params={}) def test_load_temp_staging_raw_table(mock_aws, mock_executor, aws_config_mock): """Test load_temp_staging_raw_table method.""" mock_executor.load_temp_staging_raw_table( temp_staging_raw_table='load_temp_staging_raw_table', aws=aws_config_mock, key_dir='s3://somepath', file_pattern='test_pattern') assert mock_executor.fetchall.call_count == 1 mock_executor.fetchall.assert_any_call( SubstringMatcher( containing=['COPY INTO', 'test_db.test_schema']), params={ 'aws_key_id': 'FOOBARKEY', 'aws_secret_key': 'FOOBARSECRET', 'aws_token': 'FOOBARTOKEN', 's3_path': 's3://somepath', 'file_pattern': 'test_pattern'}) def test_clean_staging_raw_table(mock_executor): """Test clean_staging_raw_table method.""" for licensor in config.spotify_api_licensors: mock_executor.clean_staging_raw_table( temp_staging_raw_table='temp_table', staging_raw_table='staging_raw_table', date='2017-11-16', report_name='sub_30_sec_streams', licensor=licensor) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'DELETE FROM', 'test_db.test_schema.staging_raw_table', 'licensor', ]), params={ 'date': '2017-11-16', 'licensor': licensor, 'api_licensor': licensor, 'consumer_db': config.consumer_sf['db'], 'consumer_schema': config.consumer_sf['schema']}) expected = len(config.spotify_api_licensors) assert mock_executor.ex_mock.call_count == expected def test_load_staging_raw_table_for_streams( mock_executor, mock_temp_staging_raw_names): """Test load_staging_raw_table method.""" common_staging_raw_tables = {'tracks': 'spotify_tracks'} for licensor in config.spotify_api_licensors: mock_executor.load_staging_raw_table( temp_staging_raw_table='temp_table', staging_raw_table='staging_raw_table', date='2021-2-10', filename='filename', temp_staging_raw_names=mock_temp_staging_raw_names[licensor], report_name='streams', query='load_staging_raw_streams', common_staging_raw_tables=common_staging_raw_tables, licensor=licensor ) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'INSERT INTO', '{}.{}.staging_raw_table'.format( config.consumer_sf['db'], config.consumer_sf['schema']), 'from test_db.test_schema.' 'temp_staging_raw_spotify_streams', 'left join test_db.test_schema.' 'temp_staging_raw_spotify_tracks' ]), params={ 'date': '2021-2-10', 'licensor': licensor, 'tracks_table': 'spotify_tracks', 'filename': ''}) expected = len(config.spotify_api_licensors) assert mock_executor.ex_mock.call_count == expected def test_load_staging_raw_table_for_sub_30_sec_streams( mock_executor, mock_temp_staging_raw_names): """Test load_staging_raw_table method.""" common_staging_raw_tables = {'tracks': 'spotify_tracks'} for licensor in config.spotify_api_licensors: mock_executor.load_staging_raw_table( temp_staging_raw_table='temp_table', staging_raw_table='staging_raw_table', date='2021-2-10', filename='filename', temp_staging_raw_names=mock_temp_staging_raw_names[licensor], report_name='sub_30_sec_streams', query='load_staging_raw_streams', common_staging_raw_tables=common_staging_raw_tables, licensor=licensor) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'INSERT INTO', '{}.{}.staging_raw_table'.format( config.consumer_sf['db'], config.consumer_sf['schema']), 'from test_db.test_schema.' 'temp_staging_raw_spotify_sub_30_sec_streams', 'spotify_tracks' ]), params={ 'date': '2021-2-10', 'licensor': licensor, 'tracks_table': 'spotify_tracks', 'filename': ''}) expected = len(config.spotify_api_licensors) assert mock_executor.ex_mock.call_count == expected def test_load_staging_raw_table_for_aggregated_streams( mock_executor, mock_temp_staging_raw_names): """Test load_staging_raw_table method.""" common_staging_raw_tables = {'users': 'spotify_users', 'tracks': 'spotify_tracks'} for licensor in config.spotify_api_licensors: mock_executor.load_staging_raw_table( temp_staging_raw_table='temp_table', staging_raw_table='staging_raw_table', date='2017-11-16', filename='filename', temp_staging_raw_names=mock_temp_staging_raw_names[licensor], report_name='aggregated_streams', query='load_staging_raw_aggregated_streams', common_staging_raw_tables=common_staging_raw_tables, licensor=licensor) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'INSERT INTO', 'content:shares', 'test_db.test_schema.staging_raw_table', 'from test_db.test_schema.', 'from test_db.test_schema.' 'temp_staging_raw_spotify_aggregated_streams', 'LEFT JOIN', 'spotify_tracks' ]), params={ 'filename': 'aggregated_streams_2017-11-16_{}.gz'.format( licensor), 'temp_users_staging_raw_table': ( 'temp_staging_raw_spotify_users_20171116_{}'.format( licensor)), 'temp_tracks_staging_raw_table': ( 'temp_staging_raw_spotify_tracks_20171116_{}'.format( licensor)), 'licensor': licensor, 'date': '2017-11-16', 'consumer_db': config.consumer_sf['db'], 'consumer_schema': config.consumer_sf['schema']}) expected = len(config.spotify_api_licensors) assert mock_executor.ex_mock.call_count == expected def test_load_staging_fact_table(mock_executor_theorchard): """Test load_staging_fact_table method.""" mock_executor_theorchard.load_staging_fact_table('2017-11-16') assert mock_executor_theorchard.ex_mock.call_count == 1 mock_executor_theorchard.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.staging_fact_analytics_' 'spotify_theorchard_streams_20171116', '{}.{}.staging_raw_spotify_v2'.format( config.consumer_sf['db'], config.consumer_sf['schema'])]), params={ 'reportdate': '2017-11-16', 'storeid': 286, 'feedid': 1, 'licensor': 'theorchard'}) def test_delete_from_fact_table(mock_executor_theorchard): """Test delete_from_fact_table method.""" mock_executor_theorchard.delete_from_fact_table('2017-11-16') assert mock_executor_theorchard.ex_mock.call_count == 1 mock_executor_theorchard.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'DELETE FROM', 'test_db.test_schema.fact_analytics']), params={ 'reportdate': '2017-11-16', 'storeid': 286, 'feedid': 1, 'date': '2017-11-16', 'staging_raw_table': 'staging_raw_spotify_v2', 'licensor': 'theorchard' }) def test_load_fact_data(mock_executor): """Test load_fact_data method.""" mock_executor.load_fact_data('2017-11-16') assert mock_executor.ex_mock.call_count == 1 mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'INSERT INTO', 'test_db.test_schema.fact_analytics', 'test_db.test_schema.staging_fact_analytics_spotify_20171116', 'test_db.test_schema.staging_raw_spotify_v2']), params={'reportdate': '2017-11-16'}) def test_delete_from_fact_error_table(mock_executor_theorchard): """Test delete_from_fact_error_table method.""" mock_executor_theorchard.delete_from_fact_error_table('2017-11-16') assert mock_executor_theorchard.ex_mock.call_count == 1 mock_executor_theorchard.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'DELETE FROM', 'test_db.test_schema.fact_analytics_error']), params={ 'reportdate': '2017-11-16', 'storeid': 286, 'feedid': 1, 'date': '2017-11-16', 'staging_raw_table': 'staging_raw_spotify_v2', 'licensor': 'theorchard' }) def test_load_fact_error_data(mock_executor_theorchard): """Test load_fact_error_data method.""" mock_executor_theorchard.load_fact_error_data('2017-11-16') assert mock_executor_theorchard.ex_mock.call_count == 1 mock_executor_theorchard.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.fact_analytics_error', '{}.{}.staging_raw_spotify'.format( config.consumer_sf['db'], config.consumer_sf['schema']), 'test_db.test_schema.' 'staging_fact_analytics_spotify_theorchard_streams' '_20171116']), params={ 'reportdate': '2017-11-16', 'storeid': 286, 'feedid': 1, 'licensor': 'theorchard'}) def test_update_dimension_table(mock_executor_theorchard): """Test update_dimension_table method.""" mock_executor_theorchard.fetchone = MagicMock() mock_executor_theorchard.update_dimension_table( '2017-11-16', 'dim_sourcetype') assert mock_executor_theorchard.fetchone.call_count == 1 mock_executor_theorchard.fetchone.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.dim_sourcetype']), params={ 'date': '2017-11-16', 'storeid': 286, 'feedid': 1, 'licensor': 'theorchard'}, dict_cursor=True) def test_create_transitional_temp_staging_raw_table( mock_executor, mock_common_tables): """Test create_transition_temp_staging_raw_table method.""" for record in mock_common_tables: mock_executor.create_transitional_temp_staging_raw_table( record.report_name, record.transitional_temp_table) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'CREATE OR REPLACE TRANSIENT', 'test_db.test_schema', record.transitional_temp_table]), params={}) assert mock_executor.ex_mock.call_count == 1 def test_load_transitional_common_tables( mock_executor, mock_temp_staging_raw_names, mock_common_tables): """Test load_transitional_common_tables method.""" for record, licensor in product( mock_common_tables, mock_temp_staging_raw_names): mock_executor.load_transitional_common_tables( '2017-11-16', mock_temp_staging_raw_names[licensor][record.report_name], record.staging_table, record.transitional_temp_table, record.report_name, licensor) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'INSERT INTO', 'test_db.test_schema.{}'.format( record.transitional_temp_table), 'test_db.test_schema.{}'.format(record.staging_table), ]), params={ 'date': '2017-11-16', 'licensor': licensor}) expected = len(config.spotify_api_licensors)*len(['tracks']) assert mock_executor.ex_mock.call_count == expected def test_load_common_staging_raw_table(mock_executor, mock_common_tables): """Test load_transitional_common_tables method.""" for record, licensor in product( mock_common_tables, config.spotify_api_licensors): mock_executor.load_common_staging_raw_table( '2017-11-16', record.transitional_temp_table, record.staging_table, record.report_name, licensor) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'MERGE INTO', 'test_db.test_schema.{}'.format(record.staging_table), 'test_db.test_schema.{}'.format( record.transitional_temp_table), 'WHEN NOT MATCHED THEN', 'INSERT' ]), params={ 'date': '2017-11-16', 'licensor': licensor}) expected = len(config.spotify_api_licensors) * len(['tracks']) assert mock_executor.ex_mock.call_count == expected