"""Unit tests for Pandora Snowflake SQL executor.""" from unittest.mock import MagicMock from unittest.mock import patch import pytest from snowflake import connector from feed_ingestion.flows.pandora import config from feed_ingestion.flows.pandora import util from feed_ingestion.flows.pandora.snowflake_executor import PandoraSMEFA from feed_ingestion.flows.pandora.snowflake_executor import PandoraTheOrchardFA from tests.conftest import SubstringMatcher @pytest.fixture(autouse=True) def patch_connector(monkeypatch): """Patch Snowflake connector to prevent real calls.""" connect_mock = MagicMock() monkeypatch.setattr(connector, 'connect', connect_mock) @pytest.fixture def mock_executor(sf_config_mock, monkeypatch): """Yield Theorchard executor context.""" connect_mock = MagicMock() monkeypatch.setattr(connector, 'connect', connect_mock) executor = PandoraTheOrchardFA(sf_config_mock) with patch.object(executor, 'execute', wraps=executor.execute) as \ executor.ex_mock: with patch.object(executor, 'fetchall', wraps=executor.fetchall) as \ executor.fa_mock: yield executor @pytest.fixture def mock_executor_sme(sf_config_mock, monkeypatch): """Yield executor context.""" connect_mock = MagicMock() monkeypatch.setattr(connector, 'connect', connect_mock) executor = PandoraSMEFA(sf_config_mock) with patch.object(executor, 'execute', wraps=executor.execute) as \ executor.ex_mock: yield executor @pytest.mark.parametrize('date, sql_texts', [ ('2016-10-24', ['subscription_type VARCHAR(100)', 'source_detail VARCHAR(10)']), ('2016-10-25', ['-- subscription_type VARCHAR(100)', '-- source_detail VARCHAR(10)']), ('2016-11-14', ['-- subscription_type VARCHAR(100)', '-- source_detail VARCHAR(10)']), ('2016-10-10', ['subscription_type VARCHAR(100)', 'source_detail VARCHAR(10)']), ('2021-09-02', ['zipcode VARCHAR']), ]) def test_create_streams_temp_staging_raw_table(mock_executor, date, sql_texts): """Test create_streams_temp_staging_raw_table method.""" mock_executor.create_streams_temp_staging_raw_table( 'temp_streams_table', date_for_sqlloader=date) assert mock_executor.ex_mock.call_count == 1 containing = ['test_db.test_schema', 'temp_streams_table'] + sql_texts mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=containing), params={}) @pytest.mark.parametrize('date, sql_text', [ ('2016-10-24', 'artist_uri VARCHAR(1000),'), ('2016-10-25', '-- artist_uri VARCHAR(1000),'), ('2016-11-14', '-- artist_uri VARCHAR(1000),'), ('2016-10-10', 'artist_uri VARCHAR(1000),'), ]) def test_create_metadata_temp_staging_raw_table(mock_executor, date, sql_text): """Test create_metadata_temp_staging_raw_table method.""" mock_executor.create_metadata_temp_staging_raw_table( 'temp_metadata_table', date_for_sqlloader=date) assert mock_executor.ex_mock.call_count == 1 mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=['test_db.test_schema', 'temp_metadata_table', sql_text]), params={}) @pytest.mark.parametrize('kwargs, on_error_action', [ ({}, 'ABORT_STATEMENT'), ({'error_limit': None}, 'ABORT_STATEMENT'), ({'error_limit': 10}, 'SKIP_FILE_10'), ({'error_limit': 25}, 'SKIP_FILE_25') ]) def test_load_temp_staging_raw_table( mock_aws, mock_executor, kwargs, on_error_action): """Test load_temp_staging_raw_table method.""" mock_executor.load_temp_staging_raw_table( 'temp_streams_table_US', None, 's3://somepath', **kwargs) assert mock_executor.fa_mock.call_count == 1 mock_executor.fa_mock.assert_any_call( SubstringMatcher( containing=['test_db.test_schema', 'temp_streams_table_US', on_error_action]), params={'aws_key_id': 'FOOBARKEY', 'aws_secret_key': 'FOOBARSECRET', 'aws_token': 'FOOBARTOKEN', 's3_path': 's3://somepath'}) @pytest.mark.parametrize('kwargs, error_on_column_count_mismatch', [ ({'error_on_column_count_mismatch': ''}, 'TRUE'), ({'error_on_column_count_mismatch': 'TRUE'}, 'TRUE'), ({'error_on_column_count_mismatch': 'anyThiNGElse'}, 'TRUE'), ({'error_on_column_count_mismatch': 'false'}, 'FALSE') ]) def test_load_temp_staging_raw_table_on_column_count_mismatch( mock_aws, mock_executor, kwargs, error_on_column_count_mismatch): """Test load_temp_staging_raw_table method.""" mock_executor.load_temp_staging_raw_table( 'temp_streams_table_US', None, 's3://somepath', **kwargs) assert mock_executor.fa_mock.call_count == 1 mock_executor.fa_mock.assert_any_call( SubstringMatcher( containing=['test_db.test_schema', 'temp_streams_table_US', error_on_column_count_mismatch]), params={'aws_key_id': 'FOOBARKEY', 'aws_secret_key': 'FOOBARSECRET', 'aws_token': 'FOOBARTOKEN', 's3_path': 's3://somepath'}) def test_clean_staging_raw_table_theorchard(mock_executor): """Test clean_staging_raw_table method.""" mock_executor.clean_staging_raw_table('staging_raw_pandora', '2017-01-01') assert mock_executor.ex_mock.call_count == 1 mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=['test_db.test_schema.staging_raw_pandora']), params={'date': '2017-01-01', 'licensor': 'theorchard'}) def test_clean_staging_raw_table_sme(mock_executor_sme): """Test clean_staging_raw_table method.""" mock_executor_sme.clean_staging_raw_table( 'staging_raw_pandora', '2017-01-01') assert mock_executor_sme.ex_mock.call_count == 1 mock_executor_sme.ex_mock.assert_any_call( SubstringMatcher( containing=['test_db.test_schema.staging_raw_pandora']), params={'date': '2017-01-01', 'licensor': 'sme'}) def test_load_staging_raw_table_theorchard(mock_executor): """Test load_staging_raw_table method.""" mock_executor.load_staging_raw_table( '2017-01-01', '2017-01-01T14:34:51', 'streamsUS.bz2', 'staging_raw_pandora', 'temp_streams_US_table', 'temp_metadata_table') assert mock_executor.ex_mock.call_count == 1 mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.staging_raw_pandora', 'test_db.test_schema.temp_streams_US_table', 'test_db.test_schema.temp_metadata_table']), params={ 'date': '2017-01-01', 'processed_datetime': '2017-01-01T14:34:51', 'filename': 'streamsUS.bz2', 'licensor': 'theorchard'}) def test_load_staging_raw_table_sme(mock_executor_sme): """Test load_staging_raw_table method.""" mock_executor_sme.load_staging_raw_table( '2017-01-01', '2017-01-01T14:34:51', 'streamsUS.bz2', 'staging_raw_pandora', 'temp_streams_US_table', 'temp_metadata_table') assert mock_executor_sme.ex_mock.call_count == 1 mock_executor_sme.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.staging_raw_pandora', 'test_db.test_schema.temp_streams_US_table', 'test_db.test_schema.temp_metadata_table']), params={ 'date': '2017-01-01', 'processed_datetime': '2017-01-01T14:34:51', 'filename': 'streamsUS.bz2', 'licensor': 'sme'}) def test_update_dimension_table(mock_executor): """Test update_dimension_table method.""" mock_executor.fetchone = MagicMock() mock_executor.update_dimension_table('2017-11-16', 'dim_user') assert mock_executor.fetchone.call_count == 1 mock_executor.fetchone.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.dim_user', 'test_db.test_schema.staging_raw_pandora', 'test_db.test_schema.dim_licensor']), params={ 'date': '2017-11-16', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'}, dict_cursor=True) def test_update_dimension_table_sme(mock_executor_sme): """Test update_dimension_table method.""" mock_executor_sme.fetchone = MagicMock() mock_executor_sme.update_dimension_table('2017-11-16', 'dim_user') assert mock_executor_sme.fetchone.call_count == 1 mock_executor_sme.fetchone.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.dim_user', 'test_db.test_schema.staging_raw_pandora', 'test_db.test_schema.dim_licensor']), params={ 'date': '2017-11-16', 'storeid': 708, 'feedid': 3, 'licensor': 'sme'}, dict_cursor=True) def test_load_staging_fact_table_theorchard(mock_executor): """Test load_staging_fact_table method.""" mock_executor.load_staging_fact_table('2017-01-01') assert mock_executor.ex_mock.call_count == 1 mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.' 'staging_fact_analytics_pandora_theorchard_20170101', 'test_db.test_schema.staging_raw_pandora', util.build_cases_by_country(config.country_mappings), util.build_cases_by_country(config.currencyid_mappings) ]), params={ 'reportdate': '2017-01-01', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'}) def test_load_staging_fact_table_sme(mock_executor_sme): """Test load_staging_fact_table method.""" mock_executor_sme.load_staging_fact_table('2017-01-01') assert mock_executor_sme.ex_mock.call_count == 1 mock_executor_sme.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.' 'staging_fact_analytics_pandora_sme_20170101', 'test_db.test_schema.staging_raw_pandora', util.build_cases_by_country(config.country_mappings), util.build_cases_by_country(config.currencyid_mappings) ]), params={ 'reportdate': '2017-01-01', 'storeid': 708, 'feedid': 3, 'licensor': 'sme'}) def test_load_fact_error_data_theorchard(mock_executor): """Test load_fact_error_data method.""" mock_executor.load_fact_error_data('2017-01-01') assert mock_executor.ex_mock.call_count == 1 print(mock_executor.ex_mock.mock_calls[0][1][0]) mock_executor.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.' 'staging_fact_analytics_pandora_theorchard_20170101', 'test_db.test_schema.staging_raw_pandora', 'test_db.test_schema.fact_analytics_error', util.build_cases_by_country(config.country_mappings), util.build_cases_by_country(config.currencyid_mappings) ]), params={ 'reportdate': '2017-01-01', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'}) def test_load_fact_error_data_sme(mock_executor_sme): """Test load_fact_error_data method.""" mock_executor_sme.load_fact_error_data('2017-01-01') assert mock_executor_sme.ex_mock.call_count == 1 mock_executor_sme.ex_mock.assert_any_call( SubstringMatcher( containing=[ 'test_db.test_schema.' 'staging_fact_analytics_pandora_sme_20170101', 'test_db.test_schema.staging_raw_pandora', 'test_db.test_schema.fact_analytics_error', util.build_cases_by_country(config.country_mappings), util.build_cases_by_country(config.currencyid_mappings) ]), params={ 'reportdate': '2017-01-01', 'storeid': 708, 'feedid': 3, 'licensor': 'sme'})