"""Unit tests for GFKSL class overriding StageLoader component.""" from unittest import mock from unittest.mock import MagicMock import freezegun from freezegun.api import FakeDatetime from feed_ingestion.flows.gfk.stage_loader import GFKSL def test_get_release_year_and_week(): """Test get_release_year_and_week method.""" executor = MagicMock() sql_loader = MagicMock() stage_loader = GFKSL(executor, sql_loader) executor.fetchone_query.return_value = ('201801',) stage_loader.get_release_year_and_week('2018-01-01', 'mystage') executor.fetchone_query.assert_called_with( sql_loader, 'get_release_year_and_week', params={ 'db': mock.ANY, 'schema': mock.ANY, 'stage_name': 'mystage', 'date': '2018-01-01'}) @freezegun.freeze_time('2018-01-01') def test_load_staging_raw_table(monkeypatch): """Test delete_from_fact_table method.""" executor = MagicMock() sql_loader = MagicMock() resolve_sql_loader_and_execute_mock = MagicMock() get_release_year_and_week_mock = MagicMock( return_value=('2018-01-01', '2018-01-06')) monkeypatch.setattr( GFKSL, 'resolve_sql_loader_and_execute', resolve_sql_loader_and_execute_mock) monkeypatch.setattr( GFKSL, 'get_release_year_and_week', get_release_year_and_week_mock) stage_loader = GFKSL(executor, sql_loader) stage_name = 'my_stage' stage_raw_table = 'my_staging_raw_table' date = '2018-01-01' source_files_dict = { 'files': [ { 'file_name': 'membran.gz', 'found': True, 'file_size': 1000 } ] } stage_loader.load_staging_raw_table( stage_raw_table, source_files_dict, date, stage_name) assert resolve_sql_loader_and_execute_mock.call_count == 1 resolve_sql_loader_and_execute_mock.assert_called_with( 'load_staging_raw', params={ 'db': mock.ANY, 'schema': mock.ANY, 'stage': 'my_stage', 'staging_raw_table': 'my_staging_raw_table', 'file_name': 'membran.gz', 'file_size': 1000, 'download_date': '2018-01-01', 'ingestion_time': FakeDatetime(2018, 1, 1, 0, 0), 'first_day_of_release_week': '2018-01-01', 'last_day_of_release_week': '2018-01-06'})