"""Script for executing the lambda locally against AWS Dev.""" import os from time import sleep from src import app as index os.environ['SNS_FINAL_STATUS_TOPIC_ARN'] = 'arn:aws:sns:us-east-1:103233932089:dev-asset_transcoder_final_status' os.environ['LAMBDA_TASK_ROOT'] = '/usr/local' success_event = { 'bucket': 'dev-orcd-asset-transcoder-input', 'key': '81a0289f_d674_4aab_ab95_333dc61fb331.wav', 'file_type': 'wav', 'config': { 'output_bucket_name': 'dev-orcd-podcast-output-assets', 'preview_bucket_name': 'dev-orcd-podcast-output-assets', 'elastic_transcoder_id': '1581101525157-xjcv85' } } not_found_event = { 'bucket': 'dev-orcd-asset-transcoder-input', 'key': '00392262_1dbe_4a14_bf1d_b649f9984ecc.wav', 'file_type': 'wav', 'config': { 'output_bucket_name': 'dev-orcd-podcast-output-assets', 'preview_bucket_name': 'dev-orcd-podcast-output-assets', 'elastic_transcoder_id': '1581101525157-xjcv85' } } invalid_transcoder_id = { 'bucket': 'dev-orcd-asset-transcoder-input', 'key': '81a0289f_d674_4aab_ab95_333dc61fb331.wav', 'file_type': 'wav', 'config': { 'output_bucket_name': 'dev-orcd-podcast-output-assets', 'preview_bucket_name': 'dev-orcd-podcast-output-assets', 'elastic_transcoder_id': 'invalid_id' } } nonexistant_transcoder = { 'bucket': 'dev-orcd-asset-transcoder-input', 'key': '81a0289f_d674_4aab_ab95_333dc61fb331.wav', 'file_type': 'wav', 'config': { 'output_bucket_name': 'dev-orcd-podcast-output-assets', 'preview_bucket_name': 'dev-orcd-podcast-output-assets', 'elastic_transcoder_id': '0081101525157-xjcv85' } } invalid_event = { 'bucket': 'dev-orcd-asset-transcoder-input', 'key': '81a0289f_d674_4aab_ab95_333dc61fb331.wav', 'config': { 'output_bucket_name': 'dev-orcd-podcast-output-assets', 'preview_bucket_name': 'dev-orcd-podcast-output-assets', 'elastic_transcoder_id': '1581101525157-xjcv85' } } def test_event(lambda_module, event): """Method for testing the handler function with a test event.""" try: lambda_module.handler(event, None) except Exception: pass def test_binary_not_found(lambda_module): """Method for testing error handling when the binary is not found.""" del os.environ['LAMBDA_TASK_ROOT'] lambda_module.handler(success_event, None) if __name__ == '__main__': for event in [success_event, not_found_event, invalid_transcoder_id, nonexistant_transcoder, invalid_event]: test_event(index, event) sleep(1) test_binary_not_found(index)