"""Tests for processing module.""" import json from unittest import mock import ffmpy from flexmock import flexmock import pytest from transcoding import config from transcoding.constants import exceptions as transcoding_exceptions from transcoding.constants import transcoding from transcoding.logic import storage from transcoding.logic import transcoding_logic from transcoding.utils import file_utils @pytest.fixture def ffmpeg_cli_config_mapping(): """Fixture for get_ffmpeg_output_config input/output values.""" return { 'input_transcoding_job': { 'input_bucket': 'dev-mediainfo-lambda-samples', 'input_key': 'b6ad954c-5ffb-47da-9051-00a6db0d2c65.wav', 'output_key': '08d73aa0-99a5-4d6f-abcc-d6245c20a7a5.wav', 'bit_rate': 1519552, 'container': 'wave', 'channels': 2, 'codec': 'pcm', 'sample_rate': 92160, 'bit_depth': 24, }, 'expected_cli_command': ( '-acodec pcm_s24le -loglevel debug -ar 92160 -ac 2' ' -b:a 1519552'), 'expected_strip_nonaudio_cli_command': ( '-acodec pcm_s24le -loglevel debug -ar 92160 -ac 2' ' -b:a 1519552 -map 0:a') } @pytest.fixture def message_mock(ffmpeg_cli_config_mapping): """Fixture for Message mocking.""" transcoding_job = ffmpeg_cli_config_mapping['input_transcoding_job'] json_encoded_message = json.dumps(transcoding_job) message = mock.MagicMock(body=json_encoded_message) return { 'input_message': message, 'expected_body': ffmpeg_cli_config_mapping['input_transcoding_job'] } @pytest.fixture def transcode_asset_result_mocks(): """Fixture for mocking result calls of utils methods in transcode_asset.""" return { 'input_url': 'input_s3_url', 'output_path': 'tmp/worker_1/08d73aa0-99a5-4d6f-abcc-d6245c20a7a5.wav', 'ffmpeg_conf': '-acodec pcm_s24le -loglevel debug -ar 92160 -ac 2 ' '-b:a 1519552', } def test_generate_output_path(ffmpeg_cli_config_mapping): """Test for getting topic from order data.""" # data preparation filename = ffmpeg_cli_config_mapping['input_transcoding_job']['output_key'] tmp_folder = config.WORKER_TMP_DIRECTORY_PATH.format(worker_id=1) expected_path = transcoding.OUTPUT_FILENAME_PATTERN.format( output_folder=tmp_folder, filename=filename) # test functional call result_path = transcoding_logic._generate_output_path(tmp_folder, filename) # checking assert result_path == expected_path def test_get_ffmpeg_output_config_strip_nonaudio_with_ff_on( ffmpeg_cli_config_mapping): """Test for getting ffmpeg output cli configuration.""" mapping_key = 'expected_strip_nonaudio_cli_command' expected_result = ffmpeg_cli_config_mapping[mapping_key] # test functional call result_configuration = transcoding_logic.get_ffmpeg_output_config( ffmpeg_cli_config_mapping['input_transcoding_job'] ) # checking assert result_configuration == expected_result def test_transcode_asset_success(message_mock, transcode_asset_result_mocks): """Test for success call of transcode_asset.""" # data preparation worker_id = 1 tmp_path = config.WORKER_TMP_DIRECTORY_PATH.format(worker_id=worker_id) ffmpeg_mock = mock.MagicMock() ffmpeg_mock.run.return_value = (b'', b'') # mocking (flexmock(transcoding_logic) .should_receive('_generate_output_path') .with_args(tmp_path, message_mock['expected_body']['output_key']) .and_return(transcode_asset_result_mocks['output_path'])) (flexmock(transcoding_logic) .should_receive('get_ffmpeg_output_config') .with_args(message_mock['expected_body']) .and_return(transcode_asset_result_mocks['ffmpeg_conf'])) (flexmock(storage) .should_receive('generate_url_for_uploaded_asset') .with_args( bucket=message_mock['expected_body']['input_bucket'], key=message_mock['expected_body']['input_key']) .and_return(transcode_asset_result_mocks['input_url'])) (flexmock(file_utils) .should_receive('clean_directory') .with_args(tmp_path)) (flexmock(ffmpy) .should_receive('FFmpeg') .with_args( executable=config.FFMPEG_PATH, inputs={transcode_asset_result_mocks['input_url']: config.FFMPEG_INPUT_OPTIONS}, outputs={ transcode_asset_result_mocks['output_path']: transcode_asset_result_mocks['ffmpeg_conf'] }) .and_return(ffmpeg_mock)) # test function call result_path = transcoding_logic.transcode_asset( message_mock['input_message'], worker_id) # checking assert result_path == transcode_asset_result_mocks['output_path'] assert ffmpeg_mock.run.called def test_transcode_asset_runtime_error( message_mock, transcode_asset_result_mocks): """Test for failure call of transcode_asset.""" # data preparation worker_id = 1 tmp_path = config.WORKER_TMP_DIRECTORY_PATH.format(worker_id=worker_id) expected_exception = ( ffmpy.FFRuntimeError( cmd=['ffpeg'], exit_code=1, stdout=None, stderr=None)) ffmpeg_mock = mock.MagicMock() ffmpeg_mock.run.side_effect = expected_exception # mocking (flexmock(transcoding_logic) .should_receive('_generate_output_path') .with_args(tmp_path, message_mock['expected_body']['output_key']) .and_return(transcode_asset_result_mocks['output_path'])) (flexmock(transcoding_logic) .should_receive('get_ffmpeg_output_config') .with_args(message_mock['expected_body']) .and_return(transcode_asset_result_mocks['ffmpeg_conf'])) (flexmock(storage) .should_receive('generate_url_for_uploaded_asset') .with_args( bucket=message_mock['expected_body']['input_bucket'], key=message_mock['expected_body']['input_key']) .and_return(transcode_asset_result_mocks['input_url'])) (flexmock(file_utils) .should_receive('clean_directory') .with_args(tmp_path)) (flexmock(ffmpy) .should_receive('FFmpeg') .with_args( executable=config.FFMPEG_PATH, inputs={transcode_asset_result_mocks['input_url']: config.FFMPEG_INPUT_OPTIONS}, outputs={ transcode_asset_result_mocks['output_path']: transcode_asset_result_mocks['ffmpeg_conf'] }) .and_return(ffmpeg_mock)) # test function call and checking with pytest.raises(transcoding_exceptions.TranscodingRetryableError): transcoding_logic.transcode_asset( message_mock['input_message'], worker_id) def test_transcode_asset_not_found_error( message_mock, transcode_asset_result_mocks): """Test for failure call of transcode_asset.""" # data preparation worker_id = 1 tmp_path = config.WORKER_TMP_DIRECTORY_PATH.format(worker_id=worker_id) expected_exception = ffmpy.FFExecutableNotFoundError() ffmpeg_mock = mock.MagicMock() ffmpeg_mock.run.side_effect = expected_exception # mocking (flexmock(transcoding_logic) .should_receive('_generate_output_path') .with_args(tmp_path, message_mock['expected_body']['output_key']) .and_return(transcode_asset_result_mocks['output_path'])) (flexmock(transcoding_logic) .should_receive('get_ffmpeg_output_config') .with_args(message_mock['expected_body']) .and_return(transcode_asset_result_mocks['ffmpeg_conf'])) (flexmock(storage) .should_receive('generate_url_for_uploaded_asset') .with_args( bucket=message_mock['expected_body']['input_bucket'], key=message_mock['expected_body']['input_key']) .and_return(transcode_asset_result_mocks['input_url'])) (flexmock(file_utils) .should_receive('clean_directory') .with_args(tmp_path)) (flexmock(ffmpy) .should_receive('FFmpeg') .with_args( executable=config.FFMPEG_PATH, inputs={transcode_asset_result_mocks['input_url']: config.FFMPEG_INPUT_OPTIONS}, outputs={ transcode_asset_result_mocks['output_path']: transcode_asset_result_mocks['ffmpeg_conf'] }) .and_return(ffmpeg_mock)) # test function call and checking with pytest.raises(transcoding_exceptions.TranscodingFatalError): transcoding_logic.transcode_asset( message_mock['input_message'], worker_id)