"""Lambda functional test module.""" import string import csv from unittest.mock import mock_open, patch import unittest.mock as mock import pytest import numpy as np import pandas as pd from src import app from src.constants import track_fields as field @pytest.mark.parametrize( 'input_value, expected_result', [ pytest.param(np.nan, True, id='Nan case'), pytest.param('hello', False, id='valid string'), ] ) def test_is_null(input_value, expected_result): """Test is_null function with multiple test cases.""" result = app.is_null(input_value) assert result == expected_result @pytest.mark.parametrize( 'input_value, length, expected_output', [ pytest.param('short', 10, 'short', id='No trim needed'), pytest.param('this is a longer string', 10, 'this is a ', id='Trim to specified length'), pytest.param('exactlength', 11, 'exactlength', id='Exact length match'), pytest.param('a', 1, 'a', id='One character string'), pytest.param('', 5, '', id='Empty string') ] ) def test_get_length_string(input_value, length, expected_output): """Test get_length_string function with multiple test cases.""" result = app.get_length_string(input_value, length) assert result == expected_output @pytest.mark.parametrize( 'input_value, expected_output', [ pytest.param(None, '', id='None case'), pytest.param(np.nan, '', id='NaN case'), pytest.param(' Hello\tWorld ', 'HelloWorld', id='String with leading/trailing whitespace and tab'), pytest.param('NoTabOrSpace', 'NoTabOrSpace', id='String with no whitespace or tab'), pytest.param(' \t ', '', id='String with only whitespace and tabs') ] ) def test_safe_str(input_value, expected_output): """Test safe_str function with multiple test cases.""" result = app.safe_str(input_value) assert result == expected_output @pytest.mark.parametrize( 'input_value, length, expected_output', [ pytest.param(None, 5, '', id='None case'), pytest.param(np.nan, 5, '', id='NaN case'), pytest.param('Hello, World!', 5, 'Hello', id='String exceeds length'), pytest.param('Short', 10, 'Short', id='String shorter than length'), pytest.param('This is a long string', 10, 'This is a ', id='String with spaces and truncation'), pytest.param('Another example', 7, 'Another', id='String with spaces and truncation') ] ) def test_safe_trim(input_value, length, expected_output): """Test safe_trim function with multiple test cases.""" result = app.safe_trim(input_value, length) assert result == expected_output @pytest.mark.parametrize( 'track_info_df_fixture, expected_df_fixture, expected_exception', [ pytest.param('mock_track_info_df', 'mock_transform_df', None, id='valid_data'), pytest.param('mock_empty_track_info_df', 'mock_empty_transform_df', None, id='empty_track_info'), pytest.param('mock_bad_track_info_df', None, Exception, id='raises_exception'), ] ) def test_transform( track_info_df_fixture, expected_df_fixture, expected_exception, mock_hfa_orchard_track_licenses_df, mock_resubmit_orchard_track_licenses_df, mock_current_date, request ): """Test transform function with multiple test cases.""" track_info_df = request.getfixturevalue(track_info_df_fixture) if expected_exception: with pytest.raises(expected_exception): app.transform( mock_hfa_orchard_track_licenses_df, track_info_df, mock_resubmit_orchard_track_licenses_df, mock_current_date ) else: expected_df = request.getfixturevalue(expected_df_fixture) result = app.transform( mock_hfa_orchard_track_licenses_df, track_info_df, mock_resubmit_orchard_track_licenses_df, mock_current_date ) pd.testing.assert_frame_equal(result, expected_df) @pytest.mark.parametrize( 'track_fixture, track_index, publisher_index, expected_fixture, expected_fixture_index, expected_exception', [ pytest.param( 'mock_graphql_track_data', 0, 0, 'mock_track_info_json', 0, None, id='with_publisher' ), pytest.param( 'mock_graphql_track_data', 2, None, 'mock_track_info_json', 3, None, id='without_publisher' ), pytest.param( 'mock_bad_graphql_track_data', 0, None, None, None, Exception, id='invalid_track_raises_exception' ), ] ) def test_generate_flat_track_data( track_fixture, track_index, publisher_index, expected_fixture, expected_fixture_index, expected_exception, request ): """Test generate_flat_track_data function with multiple test cases.""" track_data = request.getfixturevalue(track_fixture) track = track_data[track_index] publisher = None if publisher_index is not None: publisher = track.get('publishing', {}).get('publishers', [])[publisher_index] if expected_exception: with pytest.raises(expected_exception): app.generate_flat_track_data(track, publisher) else: expected_output = request.getfixturevalue(expected_fixture)[expected_fixture_index] result = app.generate_flat_track_data(track, publisher) assert result == expected_output @pytest.mark.parametrize( 'data, expected', [ pytest.param( { field.LENGTH_MINUTE: 2, field.LENGTH_SECONDS: 30, field.INGESTION_DATE: '2022-05-13' }, True, id='all_fields_present' ), pytest.param( { field.LENGTH_MINUTE: None, field.LENGTH_SECONDS: 30, field.INGESTION_DATE: '2022-05-13' }, False, id='length_minute_none' ), pytest.param( { field.LENGTH_MINUTE: 2, field.LENGTH_SECONDS: None, field.INGESTION_DATE: '2022-05-13' }, False, id='length_seconds_none' ), pytest.param( { field.LENGTH_MINUTE: 2, field.LENGTH_SECONDS: 30, field.INGESTION_DATE: None }, False, id='ingestion_date_none' ), pytest.param( { field.LENGTH_MINUTE: None, field.LENGTH_SECONDS: None, field.INGESTION_DATE: None }, False, id='all_fields_none' ), ] ) def test_is_valid_track_data(data, expected): """Test _is_valid_track_data function with multiple test cases.""" result = app._is_valid_track_data(data) assert result is expected @pytest.mark.parametrize( 'graphql_track_data_fixture, expected_df_fixture, expected_exception', [ pytest.param('mock_graphql_track_data', 'mock_track_info_df', None, id='valid_data'), pytest.param('mock_empty_graphql_track_data', 'mock_empty_track_info_df', None, id='empty_track_info'), pytest.param('mock_bad_track_info_df', None, Exception, id='raises_exception'), ] ) def test_track_info_json_to_df( graphql_track_data_fixture, expected_df_fixture, expected_exception, request ): """Test track_info_json_to_df function.""" mock_graphql_track_data = request.getfixturevalue(graphql_track_data_fixture) if expected_exception: with pytest.raises(expected_exception): app.track_info_json_to_df(mock_graphql_track_data) else: mock_expected_track_data_df = request.getfixturevalue(expected_df_fixture) result = app.track_info_json_to_df(mock_graphql_track_data) pd.testing.assert_frame_equal(result.reset_index(drop=True), mock_expected_track_data_df.reset_index(drop=True)) def is_alphanumeric(s: str) -> bool: """Return True if the string is alphanumeric, else False.""" return all(c in string.ascii_letters + string.digits for c in s) @pytest.mark.parametrize( 'length', [5, 0, 1, 10, 100], ids=['default_length', 'zero_length', 'length_one', 'length_ten', 'length_hundred'] ) def test_generate_random_string_validity(length): """Test generate_random_string function.""" result = app.generate_random_string(length) assert isinstance(result, str) assert len(result) == length assert is_alphanumeric(result) def test_generate_random_string_is_random(): """Test 2 consecutive generate_random_string function call.""" s1 = app.generate_random_string(10) s2 = app.generate_random_string(10) assert s1 != s2, 'Expected two random strings to differ' @pytest.mark.parametrize( 'manufacturer_number, file_type, current_date, random_string, expected', [ pytest.param('12345', 'SSA', '20250101', 'XYZ99', 'Orchard_12345_SSA_20250101_XYZ99.txt', id='SSA'), pytest.param('99999', 'RGT', '20251231', 'A1B2C', 'Orchard_99999_RGT_20251231_A1B2C.txt', id='RGT'), ], ) def test_generate_file_name_variants(manufacturer_number, file_type, current_date, random_string, expected): """Test generate_file_name function.""" result = app.generate_file_name(manufacturer_number, file_type, current_date, random_string) assert result == expected def test_upload_to_s3_success(mock_dependencies): """Test upload_to_s3 function success.""" file_name = 'test_file.tsv' mock_df = mock.MagicMock(spec=pd.DataFrame) tempdir_mock = mock.MagicMock() tempdir_mock.__enter__.return_value = '/mock/tmp' mock_dependencies['temp_directory'].return_value = tempdir_mock mock_dependencies['os_path_join'].side_effect = lambda *args: '/mock/path/' + args[-1] result = app.upload_to_s3(file_name, mock_df) mock_df.to_csv.assert_called_once_with( '/mock/path/test_file.tsv', index=False, header=False, sep='\t', na_rep='', quoting=csv.QUOTE_NONE, escapechar='\\' ) mock_dependencies['upload_file_to_s3'].assert_called_once_with( 'mock-bucket', '/mock/path/test_file.tsv', 'requests/test_file.tsv' ) assert result == 'requests/test_file.tsv' def test_upload_to_s3_failure(mock_dependencies): """Test upload_to_s3 function failure.""" df = pd.DataFrame() mock_dependencies['temp_directory'].side_effect = Exception('tempdir error') with pytest.raises(Exception, match='tempdir error'): app.upload_to_s3('fail.tsv', df) @mock.patch('src.app.generate_random_string', return_value='abc123') @mock.patch('src.app.generate_file_name', return_value='generated_file.tsv') @mock.patch('src.app.upload_to_s3', return_value='requests/generated_file.tsv') @mock.patch('src.app.logger') def test_generate_and_upload_file_success( mock_logger, mock_upload_to_s3, mock_generate_file_name, mock_generate_random_string, mock_current_date, mock_ssa_df ): """Test generate_and_upload_file function success.""" agreement_code = 'SSA' result = app.generate_and_upload_file(agreement_code, mock_ssa_df, mock_current_date) mock_generate_random_string.assert_called_once() mock_generate_file_name.assert_called_once_with('M16106', agreement_code, mock_current_date, 'abc123') mock_upload_to_s3.assert_called_once_with('generated_file.tsv', mock_ssa_df) mock_logger.assert_not_called() assert result == { 'file_name': 'generated_file.tsv', 's3_file_path': 'requests/generated_file.tsv' } @mock.patch('src.app.generate_random_string', return_value='abc123') @mock.patch('src.app.generate_file_name', return_value='generated_file.tsv') @mock.patch('src.app.upload_to_s3', side_effect=Exception('Upload failed')) @mock.patch('src.app.logger') def test_generate_and_upload_file_failure( mock_logger, mock_upload_to_s3, mock_generate_file_name, mock_generate_random_string, mock_current_date, mock_ssa_df ): """Test generate_and_upload_file function failure.""" agreement_code = 'SSA' with pytest.raises(Exception, match='Upload failed'): app.generate_and_upload_file(agreement_code, mock_ssa_df, mock_current_date) mock_generate_random_string.assert_called_once() mock_generate_file_name.assert_called_once_with('M16106', agreement_code, mock_current_date, 'abc123') mock_upload_to_s3.assert_called_once_with('generated_file.tsv', mock_ssa_df) mock_logger.exception.assert_called_once() @mock.patch('src.app.s3.object_exists', return_value=True) @mock.patch('src.app.s3.download_file_object') @mock.patch('src.app.open', new_callable=mock_open, read_data='[{"track_id": 1}, {"track_id": 2}]') @mock.patch('src.app.pd.read_json') @mock.patch('src.app.track_info_json_to_df') def test_load_generate_hfa_tmp_files_success( mock_track_info_json_to_df, mock_read_json, mock_open, mock_s3_download_file_object, mock_s3_object_exists, mock_track_info_df, mock_hfa_orchard_track_licenses_df ): """Test load_generate_hfa_tmp_files function success.""" mock_read_json.return_value = mock_hfa_orchard_track_licenses_df mock_track_info_json_to_df.return_value = mock_track_info_df object_key_map = { app.common.HFA_ORCHARD_TRACK_LICENSES_FILE: 'license.json', app.common.PENDING_REQUEST_TRACK_IDS_DATA_FILE: 'track.json' } licenses_df, track_df = app.load_generate_hfa_tmp_files(object_key_map) assert licenses_df.equals(mock_hfa_orchard_track_licenses_df) assert track_df.equals(mock_track_info_df) @patch('src.app.s3.object_exists', return_value=False) @patch('src.app.s3.download_file_object') @patch('src.app.open', new_callable=mock.mock_open) @patch('src.app.pd.read_json') @patch('src.app.track_info_json_to_df') def test_load_generate_hfa_tmp_files_missing_s3_object( mock_track_info_json_to_df, mock_read_json, mock_open, mock_download_file_object, mock_object_exists ): """Test load_generate_hfa_tmp_files function missing s3 object failure.""" object_key_map = { app.common.HFA_ORCHARD_TRACK_LICENSES_FILE: 'missing_license.json' } with pytest.raises(Exception, match='S3 object not found'): app.load_generate_hfa_tmp_files(object_key_map) @patch('src.app.s3.object_exists', return_value=True) @patch('src.app.s3.download_file_object') @patch('src.app.open', new_callable=mock.mock_open, read_data='invalid json') @patch('src.app.pd.read_json', side_effect=ValueError('Invalid JSON')) @patch('src.app.track_info_json_to_df') def test_load_generate_hfa_tmp_files_json_parse_error( mock_track_info_json_to_df, mock_read_json, mock_open, mock_download_file_object, mock_object_exists ): """Test load_generate_hfa_tmp_files function json parse error failure.""" object_key_map = { app.common.HFA_ORCHARD_TRACK_LICENSES_FILE: 'corrupt.json' } with pytest.raises(ValueError, match='Invalid JSON'): app.load_generate_hfa_tmp_files(object_key_map)