"""Lambda test module.""" import json from unittest.mock import MagicMock import pymysql import pytest from lambdacommon.graphql.graphql import GraphQLError import config from src import app from src import constants from src.queries import graphql_queries @pytest.mark.parametrize( 'mock_cypher_result, expected_output, expected_log, side_effect, exception_type, exception_message', [ pytest.param( [{'collect(t.id)': ['t1', 't2']}], ['t1', 't2'], None, None, None, None, id='success' ), pytest.param( [], [], 'No results returned from Neo4j', None, None, None, id='no_results' ), pytest.param( [{'unexpected_key': ['t1']}], [], "Expected key 'collect(t.id)' not found", None, None, None, id='missing_key' ), pytest.param( None, [], 'Error querying Neo4j for synced tracks', Exception('Neo4j failed'), Exception, 'Neo4j failed', id='exception' ), ] ) def test_neo4j_synced_tracks( mocker, caplog, mock_cypher_result, expected_output, expected_log, side_effect, exception_type, exception_message ): """Test neo4j_synced_tracks for various response scenarios.""" track_ids = ['t1', 't2', 't3'] mock_run_cypher_query = mocker.patch('src.app.run_cypher_query', return_value=mock_cypher_result) mock_run_cypher_query.side_effect = side_effect if exception_type: with pytest.raises(exception_type, match=exception_message): app.neo4j_synced_tracks(track_ids) else: result = app.neo4j_synced_tracks(track_ids) assert result == expected_output if expected_log: assert expected_log in caplog.text @pytest.mark.parametrize( 'graphql_response, expected_output, expected_log, side_effect, exception_type, exception_message', [ pytest.param( {'tracks': [{'id': 't1', 'name': 'Track 1'}, {'id': 't2', 'name': 'Track 2'}]}, [{'id': 't1', 'name': 'Track 1'}, {'id': 't2', 'name': 'Track 2'}], None, None, None, None, id='success' ), pytest.param( {'tracks': []}, [], 'GraphQL query returned 0 tracks', None, None, None, id='no_tracks' ), pytest.param( None, [], 'Unexpected error in query_graphql_for_track_data.', GraphQLError([ { 'message': 'Something went wrong', 'path': ['tracks'], 'extensions': { 'code': 'INTERNAL_SERVER_ERROR', 'response': {'status': 500, 'body': {'error': 'internal'}}, 'exception': {'stacktrace': ['line 1', 'line 2']} } } ]), GraphQLError, 'Something went wrong', id='graphql_error' ), pytest.param( {'tracks': [{'id': 't1', 'name': 'Track 1'}]}, [], 'Unexpected error in query_graphql_for_track_data.', Exception('Unexpected error'), Exception, 'Unexpected error', id='unexpected_error' ), ] ) def test_query_graphql_for_track_data( mocker, caplog, graphql_response, expected_output, expected_log, side_effect, exception_type, exception_message ): """Test query_graphql_for_track_data for various scenarios.""" mock_execute = mocker.patch('src.app.execute_graphql_query') if side_effect: mock_execute.side_effect = side_effect else: mock_execute.return_value = graphql_response if exception_type: with pytest.raises(exception_type, match=exception_message): app.query_graphql_for_track_data(['t1', 't2']) else: result = app.query_graphql_for_track_data(['t1', 't2']) assert result == expected_output mock_execute.assert_called_once_with( query=graphql_queries.OBTAIN_PENDING_TRACK_ID_DATA, variables={'track_ids': ['t1', 't2']} ) if expected_log: assert expected_log in caplog.text @pytest.mark.parametrize( 'db_credentials, expected_output, expected_log, side_effect, exception_type, exception_message', [ pytest.param( {}, None, 'Incomplete DB credentials in config.PB_DB_CREDENTIALS', None, ValueError, 'Missing required database credentials', id='missing_credentials' ), pytest.param( {'host': 'localhost', 'user': 'user', 'password': 'password', 'database': 'db'}, [ {'track_id': '1', 'status': 'pending'}, {'track_id': '2', 'status': 'resubmit'} ], 'Executing HFA pending resubmit request query.', None, None, None, id='success' ), pytest.param( {'host': 'localhost', 'user': 'user', 'password': 'password', 'database': 'db'}, None, 'Database error while fetching HFA requests', pymysql.MySQLError('DB error'), pymysql.MySQLError, 'DB error', id='database_error' ), pytest.param( {'host': 'localhost', 'user': 'user', 'password': 'password', 'database': 'db'}, None, 'Unexpected error in get_pending_hfa_request_from_publishing', Exception('Unexpected error'), Exception, 'Unexpected error', id='unexpected_error' ), ] ) def test_get_pending_hfa_request_from_publishing( mocker, caplog, db_credentials, expected_output, expected_log, side_effect, exception_type, exception_message, ): """Test the function get_pending_hfa_request_from_publishing for different scenarios.""" mocker.patch.object(config, 'PB_DB_CREDENTIALS', db_credentials) mock_mysql_connection = mocker.patch('src.app.util.mysql_connection', MagicMock()) conn_mock = mock_mysql_connection.return_value.__enter__.return_value cursor_mock = conn_mock.cursor.return_value.__enter__.return_value cursor_mock.fetchall.return_value = expected_output cursor_mock.execute.side_effect = side_effect if exception_type: with pytest.raises(exception_type, match=exception_message): app.get_pending_hfa_request_from_publishing() else: result = app.get_pending_hfa_request_from_publishing() assert result == expected_output if expected_log: assert expected_log in caplog.text @pytest.mark.parametrize( 'all_results, synced_ids, expected_output', [ pytest.param( [{'orchard_track_id': 't1'}, {'orchard_track_id': 't2'}, {'orchard_track_id': 't3'}], ['t1', 't3'], [{'orchard_track_id': 't1'}, {'orchard_track_id': 't3'}], id='some_synced' ), pytest.param( [{'orchard_track_id': 't1'}, {'orchard_track_id': 't2'}], ['t1', 't2'], [{'orchard_track_id': 't1'}, {'orchard_track_id': 't2'}], id='all_synced' ), pytest.param( [{'orchard_track_id': 't1'}, {'orchard_track_id': 't2'}], [], [], id='none_synced' ), pytest.param( [], ['t1', 't2'], [], id='empty_all_results' ), pytest.param( [], [], [], id='both_empty' ), pytest.param( [{'orchard_track_id': 't1'}, {'orchard_track_id': 't2'}, {'orchard_track_id': 't3'}], ['t4'], [], id='no_match' ), ] ) def test_filter_synced_results(all_results, synced_ids, expected_output): """Test filter_synced_results for various scenarios.""" assert app.filter_synced_results(all_results, synced_ids) == expected_output @pytest.mark.parametrize('key, data, expected_log', [ pytest.param( 'data.json', [{'id': 1, 'name': 'track1'}, {'id': 2, 'name': 'track2'}], 'Uploading 1 item to S3: s3://my-bucket/tmp/data.json with dictionary length: 2', id='successful_upload' ), pytest.param( 'empty.json', [], 'Uploading 1 item to S3: s3://my-bucket/tmp/empty.json with dictionary length: 0', id='empty_data_upload' ) ]) def test_upload_to_s3(mocker, caplog, key, data, expected_log): """Test upload_to_s3 for various scenarios.""" mocker.patch('src.app.config.S3_BUCKET_NAME', 'my-bucket') mocker.patch('src.app.config.S3_TMP_FILE_DIR', 'tmp/') mock_s3 = mocker.patch('src.app.s3.put_object') app.upload_to_s3(key, data) expected_key = f'tmp/{key}' expected_json = json.dumps(data, ensure_ascii=False) mock_s3.assert_called_once_with( bucket='my-bucket', key=expected_key, data=expected_json ) assert expected_log in caplog.text @pytest.mark.parametrize( 'synced_ids, mock_side_effect, expected_output, expected_exception, expected_log', [ pytest.param( ['t1', 't2', 't3', 't4'], [[{'id': 't1'}], [{'id': 't2'}]], [{'id': 't1'}, {'id': 't2'}], None, ['Fetching GraphQL data for batch 1', 'Fetching GraphQL data for batch 2'], id='success_multiple_batches' ), pytest.param( ['t1', 't2'], Exception('GraphQL error'), None, Exception, ['GraphQL batch fetch failed for batch starting at index 0: GraphQL error'], id='error_first_batch' ), pytest.param( [], [], [], None, [], id='empty_input' ), ] ) def test_fetch_synced_track_data_in_batches( mocker, synced_ids, mock_side_effect, expected_output, expected_exception, expected_log, caplog ): """Test fetch_synced_track_data_in_batches for various scenarios.""" mocker.patch('src.app.config.BATCH_SIZE', 2) mock_query = mocker.patch('src.app.query_graphql_for_track_data') mock_query.side_effect = mock_side_effect if expected_exception: with pytest.raises(Exception, match='GraphQL error'): app.fetch_synced_track_data_in_batches(synced_ids) else: result = app.fetch_synced_track_data_in_batches(synced_ids) assert result == expected_output for log in expected_log: assert log in caplog.text if not synced_ids: mock_query.assert_not_called() def test_handler_success(mocker, mock_dependencies): """Test the handler when there are pending requests and all steps succeed.""" # Mock datetime mock_datetime = mocker.patch('src.app.datetime') mock_datetime.datetime.now.return_value.strftime.return_value = '20250101120000' pending_results = [ {'orchard_track_id': 't1'}, {'orchard_track_id': 't2'} ] synced_ids = ['t1'] filtered_data = [{'orchard_track_id': 't1'}] fetched_data = [{'id': 't1', 'title': 'Sample'}] # Setup return values mock_dependencies['get_pending'].return_value = pending_results mock_dependencies['neo4j_synced'].return_value = synced_ids mock_dependencies['filter_synced'].return_value = filtered_data mock_dependencies['fetch_batches'].return_value = fetched_data mock_dependencies['upload'].side_effect = lambda key, data: f'tmp/{key}' result = app.handler({}, {}) tmp_file = 'hfa_orchard_track_licenses_20250101120000.json' pending_file = 'pending_req_track_ids_data_20250101120000.json' mock_dependencies['upload'].assert_any_call(tmp_file, filtered_data) mock_dependencies['upload'].assert_any_call(pending_file, fetched_data) assert result == { constants.STATUS: constants.OK, constants.GENERATE_HFA_TMP_FILES: { constants.HFA_ORCHARD_TRACK_LICENSES_FILE: tmp_file, constants.PENDING_REQUEST_TRACK_IDS_DATA_FILE: pending_file, }, constants.GENERATE_HFA_TMP_FILES_LENGTH: 2, constants.FILES_TO_CLEANUP: [ f'tmp/{tmp_file}', f'tmp/{pending_file}' ] } def test_handler_no_pending(mocker, mock_dependencies): """Test the handler when there are no pending HFA requests.""" mock_datetime = mocker.patch('src.app.datetime') mock_datetime.datetime.now.return_value.strftime.return_value = '20250101120000' mock_dependencies['get_pending'].return_value = [] result = app.handler({}, {}) assert result == { constants.STATUS: constants.OK, constants.GENERATE_HFA_TMP_FILES: {}, constants.GENERATE_HFA_TMP_FILES_LENGTH: 0, constants.FILES_TO_CLEANUP: [] } mock_dependencies['upload'].assert_not_called() def test_handler_exception(mocker, mock_dependencies): """Test the handler when an exception occurs and is raised.""" mock_datetime = mocker.patch('src.app.datetime') mock_datetime.datetime.now.return_value.strftime.return_value = '20250101120000' mock_dependencies['get_pending'].side_effect = Exception('DB error') mock_capture = mocker.patch('src.app.capture_exception') with pytest.raises(Exception, match='DB error'): app.handler({}, {}) mock_capture.assert_called_once()