"""Tests for chunks utils.""" import pytest from src.utils import chunks def test_execute_in_chunks_call_args(mocker): """Test function decorated by execute_in_chunks receives args.""" mocked_config = mocker.patch('src.utils.chunks.common_config') mocked_config.MAX_CHUNK_SIZE = 2 mocked_func = mocker.Mock(side_effect=[['1', '2'], ['3']]) decorated_mocked_func = chunks.execute_in_chunks()(mocked_func) assert decorated_mocked_func([1, 2, 3], foo='foo') == ['1', '2', '3'] assert mocked_func.call_args_list == [ mocker.call([1, 2], foo='foo'), mocker.call([3], foo='foo'), ] def test_execute_in_chunks_handles_chunk_size(mocker): """Test function decorated by execute_in_chunks handles chunk size.""" mocked_config = mocker.patch('src.utils.chunks.common_config') mocked_config.MAX_CHUNK_SIZE = 4 mocked_func = mocker.Mock(side_effect=[['1', '2'], ['3']]) decorated_mocked_func = chunks.execute_in_chunks( max_chunk_size=2)(mocked_func) assert decorated_mocked_func([1, 2, 3]) == ['1', '2', '3'] assert mocked_func.call_args_list == [ mocker.call([1, 2]), mocker.call([3]), ] @pytest.mark.parametrize('chunk_size', (1, 2, 3, 4)) def test_execute_in_chunks(chunk_size, mocker): """Test function decorated by execute_in_chunks returns expected result.""" mocked_config = mocker.patch('src.utils.chunks.common_config') mocked_config.MAX_CHUNK_SIZE = chunk_size @chunks.execute_in_chunks() def to_str(integers): return [str(i) for i in integers] # Test with various input list lengths. for n in (0, 1, 2, 4, 8): input_list = list(range(n)) expected_output = [str(i) for i in input_list] assert to_str(input_list) == expected_output @pytest.mark.parametrize( 'input_list,expected_output', ( ([0, 1, 2, 3], {'0': 0, '1': 1, '2': 4, '3': 9}), ([], {}), ) ) @pytest.mark.parametrize('chunk_size', (1, 2, 3, 4)) def test_execute_in_chunks_dict_result( input_list, expected_output, chunk_size, mocker): """Test execute_in_chunks works with dict results.""" mocked_config = mocker.patch('src.utils.chunks.common_config') mocked_config.MAX_CHUNK_SIZE = 2 @chunks.execute_in_chunks() def to_squares_dict(integers): return {str(i): pow(i, 2) for i in integers} assert to_squares_dict(input_list) == expected_output