"""Test cases for common utility functions.""" import re from unittest import mock import pytest from constants import statuses import util @pytest.mark.parametrize('key, expected_result', [ ('filename.ext', {'filename': 'filename', 'ext': 'ext'}), ('12345', {}), ('randomstring.', {}) ]) @mock.patch('util.common_config') def test_parse_file_key(common_config, key, expected_result): """Test parse_file_key function.""" common_config.FILE_KEY_REGEXP = re.compile( r'^(?P[a-zA-Z]+)\.(?P\w{3})$') assert util.parse_file_key(key) == expected_result @pytest.mark.parametrize('key, expected_result', [ ('test/test.key', 'test.key'), ('test123.key', 'test123.key'), ('another-level/test/test-file.key', 'test-file.key'), ]) def test_get_file_name_from_key(key, expected_result): """Test get_file_name_from_key utility function.""" assert util.get_file_name_from_key(key) == expected_result @pytest.mark.parametrize('period_id, expected_periods_list', [ (1, [1, 2, 3]), (2, [1, 2, 3]), (3, [1, 2, 3]), (88, [88, 89, 90]), (89, [88, 89, 90]), (90, [88, 89, 90]), (187, [187, 188, 189]), (188, [187, 188, 189]), (189, [187, 188, 189]), (226, [226, 227, 228]), (227, [226, 227, 228]), (228, [226, 227, 228]), ]) def test_get_quarter_periods_by_period(period_id, expected_periods_list): """Test get_quarter_periods_by_period with different 4 quarters periods.""" result_list = util.get_quarter_periods_by_period(period_id) assert result_list == expected_periods_list @pytest.mark.parametrize( 'account_type, account_id, period_ids, expected_str', [ ('L', 10303, [217, 218, 219], 'L10303_217_218_219'), ('L', 10303, [217], 'L10303_217'), ('S', 10303, [217], 'S10303_217') ] ) def test_get_attachment_primary_key_value( account_type, account_id, period_ids, expected_str): """Test get_attachment_primary_key_value.""" result_str = util.get_attachment_primary_key_value( account_type, account_id, period_ids) assert result_str == expected_str @pytest.fixture def s3_event_fixture(): """S3 event fixture.""" return { 'Records': [ { 'eventVersion': '2.0', 'eventTime': '1970-01-01T00:00:00.000Z', 'requestParameters': { 'sourceIPAddress': '127.0.0.1' }, 's3': { 'configurationId': 'testConfigRule', 'object': { 'eTag': '0123456789abcdef0123456789abcdef', 'sequencer': '0A1B2C3D4E5F678901', 'key': '/testfolder/L10303_217_attac1.txt', 'size': 1024 }, 'bucket': { 'arn': 'bucketarn', 'name': 'sourcebucket', 'ownerIdentity': { 'principalId': 'EXAMPLE' } }, 's3SchemaVersion': '1.0' }, 'responseElements': { 'x-amz-id-2': '/mnopqrstuvwxyzABCDEFGH', 'x-amz-request-id': 'EXAMPLE123456789' }, 'awsRegion': 'us-east-1', 'eventName': 'ObjectCreated:Put', 'userIdentity': { 'principalId': 'EXAMPLE' }, 'eventSource': 'aws:s3' } ] } def test_s3_event_get_attachment_details_success(s3_event_fixture): """Test s3_event_get_attachment_details.""" file_name = 'attac1.txt' file_type = 'txt' account_type = 'L' label_id = '10303' period_id = '217' attachment_attrs = { 'filename': file_name, 'extension': file_type, 'account_type': account_type, 'label_id': label_id, 'period_id': period_id } expected_result = { 'file_key': '/testfolder/L10303_217_attac1.txt', 'file_name': file_name, 'original_file_name': file_name, 'file_type': file_type, 'bucket_name': 'sourcebucket', 'file_size': 1024, 'etag': '0123456789abcdef0123456789abcdef', 'account_type': account_type, 'label_id': '10303', 'period_ids': period_id, 'status': statuses.STATUS_COMPLETED, 'upload_date': '1970-01-01T00:00:00.000Z' } result = util.s3_event_get_attachment_details( s3_event_fixture, attachment_attrs) assert result == expected_result def test_s3_event_get_attachment_details_exception(s3_event_fixture): """Test s3_event_get_attachment_details in case of exception.""" del s3_event_fixture['Records'][0]['s3'] with pytest.raises(Exception): util.s3_event_get_attachment_details(s3_event_fixture, {}) @pytest.mark.parametrize('file_name, period_id, expected_result', [ ('test.csv', 217, 'test.csv(217)'), ('test.csv', 218, 'test.csv(218)'), ('test.csv', 219, 'test.csv(219)') ]) def test_format_quarter_file_name(file_name, period_id, expected_result): """Test format_quarter_file_name.""" result = util.format_quarter_file_name(file_name, period_id) assert expected_result == result