"""Unit tests for generic tasks of the FlowLoadFactMixinSF mixin.""" from unittest.mock import call from unittest.mock import MagicMock from unittest.mock import patch import pytest from snowflake import connector from feed_ingestion.tasks import load_fact_tables_tasks_sf from tests.conftest import SubstringMatcher @pytest.fixture(autouse=True) def sf_execute_mock(monkeypatch): """Yield execute.""" with MagicMock() as connect_mock: monkeypatch.setattr(connector, 'connect', connect_mock) yield connect_mock.return_value.cursor.return_value.execute @pytest.fixture(autouse=True) def activity_mock(): """Yield mock of a garcon activity object.""" with MagicMock() as activity: yield activity @pytest.fixture(autouse=True) def task_status_false_mock(): """Yield task status.""" with patch( 'feed_ingestion.tasks.load_fact_tables_tasks_sf.task_status') as \ task_status: task_status.is_completed_task.return_value = False task_status.mark_completed_task = MagicMock() yield task_status def test_create_staging_fact( sf_config_mock, activity_mock, sf_execute_mock, task_status_false_mock): """Test create_staging_fact task.""" load_fact_tables_tasks_sf.create_staging_fact( activity_mock, 'deezer_daily_theorchard', '2017-08-02', sf_config_mock) sf_execute_mock.assert_has_calls([ call( SubstringMatcher( containing=[ 'DROP TABLE IF EXISTS test_db.test_schema.' 'staging_fact_analytics_deezer_daily_theorchard_20170802' ]), None), call( SubstringMatcher( containing=[ 'CREATE TRANSIENT TABLE IF NOT EXISTS test_db.test_schema.' 'staging_fact_analytics_deezer_daily_theorchard_20170802' ]), None) ]) task_status_false_mock.mark_completed_task.assert_not_called() task_status_false_mock.is_completed_task.assert_called_once_with( 'deezer_daily_theorchard', '2017-08-02', load_fact_tables_tasks_sf.TASK_ID) def test_load_staging_fact( sf_config_mock, activity_mock, sf_execute_mock, task_status_false_mock): """Test load_staging_fact task.""" load_fact_tables_tasks_sf.load_staging_fact( activity_mock, 'spotify_theorchard_streams', '2017-08-02', sf_config_mock) sf_execute_mock.assert_called_with( SubstringMatcher( containing=[ 'INSERT INTO test_db.test_schema.staging_fact_' 'analytics_spotify_theorchard_streams_20170802', 'LEFT JOIN test_db.test_schema.dim_playlist', ]), {'licensor': 'theorchard', 'reportdate': '2017-08-02', 'feedid': 1, 'storeid': 286}) task_status_false_mock.mark_completed_task.assert_not_called() task_status_false_mock.is_completed_task.assert_called_once_with( 'spotify_theorchard_streams', '2017-08-02', load_fact_tables_tasks_sf.TASK_ID) def test_update_dim_tables(sf_config_mock, activity_mock, sf_execute_mock): """Test for update_dim_tables task.""" date = '2018-10-20' kwargs = { 'tables_to_update': [ 'dim_sourcetype', 'dim_subscriptiontype', 'dim_subscriptionpaytier', 'dim_playlist', 'dim_user' ], 'include_to_report': [ 'dim_sourcetype', 'dim_subscriptiontype', 'dim_subscriptionpaytier' ], 'sns_topic': 'test_topic' } load_fact_tables_tasks_sf.update_dim_tables( activity_mock, 'pandora_theorchard', date, sf_config_mock, kwargs=kwargs) sf_execute_mock.assert_has_calls([ call( SubstringMatcher( containing=[ 'MERGE INTO test_db.test_schema.dim_sourcetype']), {'date': '2018-10-20', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'} ), call( SubstringMatcher( containing=[ 'MERGE INTO test_db.test_schema.dim_subscriptiontype']), {'date': '2018-10-20', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'} ), call( SubstringMatcher( containing=[ 'MERGE INTO test_db.test_schema.dim_subscriptionpaytier']), {'date': '2018-10-20', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'} ), call( SubstringMatcher( containing=[ 'MERGE INTO test_db.test_schema.dim_playlist']), {'date': '2018-10-20', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'} ), call( SubstringMatcher( containing=[ 'MERGE INTO test_db.test_schema.dim_user']), {'date': '2018-10-20', 'storeid': 708, 'feedid': 3, 'licensor': 'theorchard'} ) ]) activity_mock.logger.info.assert_has_calls([ call('Updating dim_sourcetype...'), call('Updating dim_subscriptiontype...'), call('Updating dim_subscriptionpaytier...'), call('Updating dim_playlist...'), call('Updating dim_user...'), call('All the dimension tables were updated.') ]) def test_load_fact_data( sf_config_mock, activity_mock, sf_execute_mock, task_status_false_mock): """Test load_fact_data task.""" load_fact_tables_tasks_sf.load_fact_data( activity_mock, 'spotify_theorchard_streams', '2017-08-02', sf_config_mock) sf_execute_mock.assert_has_calls([ call( SubstringMatcher( containing=[ 'DELETE FROM test_db.test_schema.fact_analytics ' 'WHERE feedid']), {'storeid': 286, 'feedid': 1, 'reportdate': '2017-08-02', 'date': '2017-08-02', 'staging_raw_table': 'staging_raw_spotify_v2', 'licensor': 'theorchard'}), call( SubstringMatcher( containing=[ 'INSERT INTO test_db.test_schema.fact_analytics']), {'reportdate': '2017-08-02'}), call( SubstringMatcher( containing=[ 'DELETE FROM ' 'test_db.test_schema.fact_analytics_error ' 'WHERE feedid']), {'storeid': 286, 'feedid': 1, 'reportdate': '2017-08-02', 'date': '2017-08-02', 'staging_raw_table': 'staging_raw_spotify_v2', 'licensor': 'theorchard'}), call( SubstringMatcher( containing=[ 'INSERT INTO test_db.test_schema.fact_analytics_error' ]), {'storeid': 286, 'reportdate': '2017-08-02', 'feedid': 1, 'licensor': 'theorchard'}), call( SubstringMatcher( containing=[ 'DROP TABLE IF EXISTS test_db.test_schema.staging_fact_' 'analytics_spotify_theorchard_streams_20170802']), None) ]) task_status_false_mock.mark_completed_task.assert_called_once_with( 'spotify_theorchard_streams', '2017-08-02', load_fact_tables_tasks_sf.TASK_ID) task_status_false_mock.is_completed_task.assert_called_once_with( 'spotify_theorchard_streams', '2017-08-02', load_fact_tables_tasks_sf.TASK_ID)