"""Lambda test module.""" from datetime import datetime import pytest import config from src import app import pandas as pd from unittest.mock import mock_open from unittest.mock import patch from unittest.mock import MagicMock def isrc_data(all_fields=True): """Return isrc data.""" data = [{'isrc': 'testISRC1', 'upc': 'upc1'}, {'isrc': 'testISRC2', 'upc': 'upc2'}] if all_fields: for isrc in data: isrc['test_run_id'] = '0' isrc['test_start_time'] = '2024-12-17 16:22:47' return data def isrc_obj(isrc): """Return isrc with state.""" return {'isrc': isrc, 'upc': 'test'} class TestApp: """Test app class.""" @patch('src.app.get_days_isrcs') @patch('src.app.batch_isrcs') @patch('src.app.add_run_data') @patch('src.app.upload_to_s3') @patch('src.app.get_day') @patch('src.app.get_failed_isrcs_from_previous_days') def test_handler(self, mock_get_failed, mock_get_day, mock_upload_to_s3, mock_add_run_data, mock_batch_isrcs, mock_get_days_isrcs): """Test handler function.""" isrcs = isrc_data() mock_get_day.return_value = '20240101' mock_get_days_isrcs.return_value = isrc_data() previous_days_data = [{'isrc': 'testISRC3', 'upc': 'upc3', 'test_run_id': '0'}] mock_get_failed.return_value = previous_days_data mock_batch_isrcs.return_value = [isrcs[0], isrcs[1]] mock_upload_to_s3.return_value = 'bucket', 'key' result = app.handler(None, None) assert result == {'status': 'OK', 'total_isrcs': 3, 'num_of_batches': 2, 's3_bucket': 'bucket', 's3_key': 'key'} mock_get_day.assert_called() mock_get_days_isrcs.assert_called() mock_add_run_data.assert_called() expected_isrcs = isrc_data() + previous_days_data mock_batch_isrcs.assert_called_with(expected_isrcs, int(config.BATCH_SIZE)) mock_upload_to_s3.assert_called() @patch('src.app.get_day') def test_handler_exception(self, mock_get_day): """Test handler function with exception.""" mock_get_day.side_effect = Exception('Test exception') with pytest.raises(Exception, match='Test exception'): app.handler(None, None) @patch('src.app.SnowflakeClient') @patch('src.app.read_query_from_file') def test_get_todays_isrcs(self, mock_read_query_from_file, mock_snowflake_client): """Test get_todays_isrcs function.""" mock_instance = MagicMock() mock_snowflake_client.return_value = mock_instance mock_read_query_from_file.return_value = 'test query' mock_data_frame = pd.DataFrame({'ISRC': ['testISRC1', 'testISRC2'], 'UPC': ['upc1', 'upc2']}) mock_instance.execute_query.return_value = mock_data_frame assert app.get_days_isrcs('20240101') == isrc_data(False) def test_read_query_from_file(self): """Test read_query_from_file function.""" mock_file_content = 'query { test }' file_path = 'test_query.gql' with (patch('builtins.open', mock_open(read_data=mock_file_content)) as mock_file): content = app.read_query_from_file(file_path) mock_file.assert_called_once_with(file_path, 'r') assert content == mock_file_content @pytest.mark.parametrize( 'isrcs, expected', [([isrc_obj('1'), isrc_obj('2'), isrc_obj('3')], [[isrc_obj('1'), isrc_obj('2')], [isrc_obj('3')]]), ([isrc_obj('1'), isrc_obj('2')], [[isrc_obj('1'), isrc_obj('2')]]), ([], []) ]) def test_batch_isrcs(self, isrcs, expected): """Test batch_isrcs function.""" batch_size = 2 assert app.batch_isrcs(isrcs, batch_size) == expected def test_replace_query_placeholder(self): """Test replace_query_placeholder function.""" query = 'test query placeholder' placeholder = 'placeholder' value = 'value' assert app.replace_query_placeholder(query, placeholder, value) == ( 'test query value') @pytest.mark.parametrize(['last_id', 'expected'], [(None, 0), (1, 2)]) @patch('src.app.read_query_from_file') @patch('src.app.replace_query_placeholder') def test_get_run_id(self, mock_replace_query_placeholder, mock_read_query_from_file, last_id, expected): """Test get_run_id function.""" mock_read_query_from_file.return_value = 'test query' mock_replace_query_placeholder.return_value = 'test query' mock_snowflake_client = MagicMock() mock_data_frame = pd.DataFrame({'LAST_RUN_ID': [last_id]}) mock_snowflake_client.execute_query.return_value = mock_data_frame assert app.get_run_id(mock_snowflake_client) == expected mock_snowflake_client.execute_query.assert_called() mock_replace_query_placeholder.assert_called_with( 'test query', '', 'QA') mock_read_query_from_file.assert_called() @patch('src.app.SnowflakeClient') @patch('src.app.get_run_id') def test_add_run_data(self, mock_get_run_id, mock_snowflake_client): """Test add_run_data function.""" isrcs = isrc_data(False) mock_get_run_id.return_value = 1 date_string = '2024-12-09 12:29:07.146657' date_format = '%Y-%m-%d %H:%M:%S.%f' start_time = datetime.strptime(date_string, date_format) app.add_run_data(isrcs, start_time) assert isrcs == [{'isrc': 'testISRC1', 'upc': 'upc1', 'release_date': '2024-12-09', 'ingested_dim_track': False, 'ingested_dim_track_time': 'NULL', 'ingested_insights': False, 'ingested_insights_time': 'NULL', 'test_run_id': 1, 'test_start_time': '2024-12-09 12:29:07', 'test_end_time': 'NULL'}, {'isrc': 'testISRC2', 'upc': 'upc2', 'release_date': '2024-12-09', 'ingested_dim_track': False, 'ingested_dim_track_time': 'NULL', 'ingested_insights': False, 'ingested_insights_time': 'NULL', 'test_run_id': 1, 'test_start_time': '2024-12-09 12:29:07', 'test_end_time': 'NULL'}] def test_get_day(self): """Test get_day function.""" assert app.get_day({'override_day': '20240101'}) == '20240101' assert app.get_day({}) == datetime.now().strftime('%Y%m%d') @patch('src.app.S3Client') def test_upload_to_s3(self, mock_s3_client): """Test upload_to_s3 function.""" data = 'test data' run_id = '1' mock_instance = MagicMock() mock_s3_client.return_value = mock_instance bucket, key = app.upload_to_s3(data, run_id) assert bucket == mock_instance.bucket_name assert key == 'isrcBatchesFromSnowflake-1.json' mock_instance.put_object.assert_called() def test_get_formatted_time(self): """Test get_formatted_time function.""" assert app.get_formatted_time(None) == 'NULL' assert app.get_formatted_time(pd.NaT) == 'NULL' time_string = '2024-12-09 12:29:07' time = datetime.strptime(time_string, '%Y-%m-%d %H:%M:%S') assert app.get_formatted_time(time) == time_string def test_get_day_minus(self): """Test get_day_minus function.""" assert app.get_day_minus('20240101', 1) == '20231231' assert app.get_day_minus('20240101', 2) == '20231230' assert app.get_day_minus('20240101', 0) == '20240101' @patch('src.app.read_query_from_file') @patch('src.app.SnowflakeClient') @patch('src.app.get_day_minus') @patch('src.app.build_isrc_array') def test_get_failed_isrcs_from_previous_days(self, mock_build_isrc_array, mock_get_day_minus, mock_snowflake_client, mock_read_query_from_file): """Test get_failed_isrcs_from_previous_days function.""" day = '20240112' days_less = 1 test_run_id = 1 test_start_time = '2024-12-09 12:29:07' mock_get_day_minus.return_value = '2024005' mock_read_query_from_file.return_value = ('' ' ') mock_instance = MagicMock() mock_instance.execute_query.return_value = pd.DataFrame( [{'ISRC': 'testISRC3', 'UPC': 'upc3', 'RELEASE_DATE': datetime.strptime( '20241218', '%Y%m%d'), 'INGESTED_DIM_TRACK': False, 'INGESTED_DIM_TRACK_TIME': None, 'INGESTED_INSIGHTS': False, 'INGESTED_INSIGHTS_TIME': None, 'TEST_RUN_ID': '0'}]) mock_snowflake_client.return_value = mock_instance isrc_array = [ {'isrc': 'testISRC3', 'upc': 'upc3', 'release_date': '2024-12-18', 'test_run_id': '0', 'test_start_time': '2024-12-09 12:29:07'}] mock_build_isrc_array.return_value = isrc_array assert app.get_failed_isrcs_from_previous_days( day, days_less, test_run_id, test_start_time) == isrc_array mock_instance.execute_query.assert_called() mock_get_day_minus.assert_called() mock_instance.execute_query.assert_called_once_with( 'QA 2024005') def test_build_isrc_array(self): """Test build_isrc_array function.""" isrcs = [{'ISRC': 'testISRC1', 'UPC': 'upc1', 'RELEASE_DATE': datetime.strptime('20241218', '%Y%m%d'), 'INGESTED_DIM_TRACK': False, 'INGESTED_DIM_TRACK_TIME': None, 'INGESTED_INSIGHTS': False, 'INGESTED_INSIGHTS_TIME': None, 'TEST_RUN_ID': '0'}] test_run_id = 1 test_start_time = '2024-12-09 12:29:07' assert app.build_isrc_array(isrcs, test_run_id, test_start_time) == [ {'isrc': 'testISRC1', 'upc': 'upc1', 'release_date': '2024-12-18', 'ingested_dim_track': False, 'ingested_dim_track_time': 'NULL', 'ingested_insights': False, 'ingested_insights_time': 'NULL', 'test_run_id': 1, 'test_start_time': '2024-12-09 12:29:07', 'test_end_time': 'NULL', 'rerun': True}]