Unlock 30% off on Manual Testing Annual Plans this Holiday Season. 
Use Code: LCMYEARENDEXCL30
Use Code: LCMYEARENDEXCL30Check icon
Copied to Clipboard!
How to use lambda_client method in localstack
Best Python code snippet using localstack_python
test_index.py
Source:test_index.py 
1# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.2# SPDX-License-Identifier: Apache-2.03from unittest import TestCase4from unittest.mock import patch, MagicMock5from botocore.exceptions import ClientError6with patch("boto3.client") as boto_client_mock:7 with patch("gamekitresourcemanagement.cfn_custom_resource.send_success_response") as mock_send_success_response:8 with patch('gamekitresourcemanagement.cfn_custom_resource.send_failure_response') as mock_send_failure_response:9 from functions.main.RemoveLambdaLayersOnDelete import index10LAYER_1 = 'LambdaLayer1'11LAYER_2 = 'LambdaLayer2'12UNMATCHED_LAYER = 'UnmatchedLambdaLayer'13LAYER_VERSIONS_1 = [1, 2, 3]14LAYER_VERSIONS_2 = [1]15NEXT_MARKER = 'NextMarker'16LAYER_PREFIX = 'LambdaLayer'17PHYSICAL_RESOURCE_ID = 'delete-lambda-layers-some-hash-here'18STACK_ID = 'test_stack_id'19LOGICAL_RESOURCE_ID = 'test_logical_resource_id'20REQUEST_ID = 'test_request_id'21RESPONSE_URL = 'https://website.tld/some_url'22class TestIndex(TestCase):23 def setUp(self):24 index.lambda_client = MagicMock()25 mock_send_success_response.reset_mock()26 mock_send_failure_response.reset_mock()27 def test_removes_all_matching_lambda_layers_no_pagination(self):28 # Arrange29 event = self.get_event()30 context = self.get_context()31 index.lambda_client.list_layers.return_value = self.get_list_layers_response(32 [LAYER_1, LAYER_2, UNMATCHED_LAYER])33 index.lambda_client.list_layer_versions.side_effect = [34 self.get_list_layer_versions_response(LAYER_VERSIONS_1),35 self.get_list_layer_versions_response(LAYER_VERSIONS_2)36 ]37 # Act38 index.lambda_handler(event, context)39 # Assert40 index.lambda_client.list_layers.assert_called_once()41 self.assertEqual(index.lambda_client.list_layer_versions.call_count, 2)42 self.assertEqual(index.lambda_client.delete_layer_version.call_count, 4)43 mock_send_success_response.assert_called_once()44 mock_send_failure_response.assert_not_called()45 def test_removes_all_matching_lambda_layers_pagination(self):46 # Arrange47 event = self.get_event()48 context = self.get_context()49 index.lambda_client.list_layers.side_effect = [50 self.get_list_layers_response([LAYER_1], NEXT_MARKER),51 self.get_list_layers_response([LAYER_2, UNMATCHED_LAYER])52 ]53 # Paginate layer versions request for the first layer54 index.lambda_client.list_layer_versions.side_effect = [55 self.get_list_layer_versions_response(LAYER_VERSIONS_1[0:2], NEXT_MARKER),56 self.get_list_layer_versions_response(LAYER_VERSIONS_1[2:]),57 self.get_list_layer_versions_response(LAYER_VERSIONS_2)58 ]59 # Act60 index.lambda_handler(event, context)61 # Assert62 self.assertEqual(index.lambda_client.list_layers.call_count, 2)63 self.assertEqual(index.lambda_client.list_layer_versions.call_count, 3)64 self.assertEqual(index.lambda_client.delete_layer_version.call_count, 4)65 mock_send_success_response.assert_called_once()66 mock_send_failure_response.assert_not_called()67 def test_deletes_nothing_if_no_layers_exist(self):68 # Arrange69 event = self.get_event()70 context = self.get_context()71 index.lambda_client.list_layers.return_value = self.get_list_layers_response([])72 # Act73 index.lambda_handler(event, context)74 # Assert75 index.lambda_client.list_layers.assert_called_once()76 index.lambda_client.list_layer_versions.assert_not_called()77 index.lambda_client.delete_layer_version.assert_not_called()78 mock_send_success_response.assert_called_once()79 mock_send_failure_response.assert_not_called()80 def test_sends_failure_on_list_layers_error(self):81 # Arrange82 event = self.get_event()83 context = self.get_context()84 index.lambda_client.list_layers.side_effect = ClientError({85 'Error': {86 'Code': 500,87 'Message': 'Some Lambda Error'88 }89 }, 'ListLayers')90 # Act91 with self.assertRaises(ClientError):92 index.lambda_handler(event, context)93 # Assert94 index.lambda_client.list_layers.assert_called_once()95 index.lambda_client.list_layer_versions.assert_not_called()96 index.lambda_client.delete_layer_version.assert_not_called()97 mock_send_success_response.assert_not_called()98 mock_send_failure_response.assert_called_once()99 def test_sends_failure_on_list_layer_versions_error(self):100 # Arrange101 event = self.get_event()102 context = self.get_context()103 index.lambda_client.list_layers.return_value = self.get_list_layers_response([LAYER_1])104 index.lambda_client.list_layer_versions.side_effect = ClientError({105 'Error': {106 'Code': 500,107 'Message': 'Some Lambda Error'108 }109 }, 'ListLayerVersions')110 # Act111 with self.assertRaises(ClientError):112 index.lambda_handler(event, context)113 # Assert114 index.lambda_client.list_layers.assert_called_once()115 index.lambda_client.list_layer_versions.assert_called_once()116 index.lambda_client.delete_layer_version.assert_not_called()117 mock_send_success_response.assert_not_called()118 mock_send_failure_response.assert_called_once()119 def test_sends_failure_on_delete_layer_version_error(self):120 # Arrange121 event = self.get_event()122 context = self.get_context()123 index.lambda_client.list_layers.return_value = self.get_list_layers_response([LAYER_1])124 index.lambda_client.list_layer_versions.return_value = self.get_list_layer_versions_response([LAYER_VERSIONS_1])125 index.lambda_client.delete_layer_version.side_effect = ClientError({126 'Error': {127 'Code': 500,128 'Message': 'Some Lambda Error'129 }130 }, 'DeleteLayerVersion')131 # Act132 with self.assertRaises(ClientError):133 index.lambda_handler(event, context)134 # Assert135 index.lambda_client.list_layers.assert_called_once()136 index.lambda_client.list_layer_versions.assert_called_once()137 index.lambda_client.delete_layer_version.assert_called_once()138 mock_send_success_response.assert_not_called()139 mock_send_failure_response.assert_called_once()140 def test_ignores_non_delete_events(self):141 # Arrange142 event = self.get_event()143 event['RequestType'] = 'Create'144 context = self.get_context()145 # Act146 index.lambda_handler(event, context)147 # Assert148 index.lambda_client.list_layers.assert_not_called()149 index.lambda_client.list_layer_versions.assert_not_called()150 index.lambda_client.delete_layer_versions.assert_not_called()151 mock_send_success_response.assert_called_once()152 mock_send_failure_response.assert_not_called()153 @staticmethod154 def get_list_layers_response(layers: [str], next_marker: str = None):155 return {156 'Layers': [{'LayerName': layer} for layer in layers],157 'NextMarker': next_marker158 }159 @staticmethod160 def get_list_layer_versions_response(layer_versions: [int], next_marker: str = None):161 return {162 'LayerVersions': [{'Version': layer_version} for layer_version in layer_versions],163 'NextMarker': next_marker164 }165 @staticmethod166 def get_event():167 return {168 'RequestType': 'Delete',169 'ResourceProperties': {170 'layer_prefix': LAYER_PREFIX171 },172 'PhysicalResourceId': PHYSICAL_RESOURCE_ID,173 'StackId': STACK_ID,174 'LogicalResourceId': LOGICAL_RESOURCE_ID,175 'RequestId': REQUEST_ID,176 'ResponseURL': RESPONSE_URL177 }178 @staticmethod179 def get_context():180 context = MagicMock()...
test_start_lambda.py
Source:test_start_lambda.py
1from concurrent.futures import ThreadPoolExecutor, as_completed2from time import time3import pytest4import boto35from botocore import UNSIGNED6from botocore.config import Config7from botocore.exceptions import ClientError8from .start_lambda_api_integ_base import StartLambdaIntegBaseClass9class TestParallelRequests(StartLambdaIntegBaseClass):10 template_path = "/testdata/invoke/template.yml"11 def setUp(self):12 self.url = "http://127.0.0.1:{}".format(self.port)13 self.lambda_client = boto3.client(14 "lambda",15 endpoint_url=self.url,16 region_name="us-east-1",17 use_ssl=False,18 verify=False,19 config=Config(signature_version=UNSIGNED, read_timeout=120, retries={"max_attempts": 0}),20 )21 @pytest.mark.flaky(reruns=3)22 @pytest.mark.timeout(timeout=300, method="thread")23 def test_same_endpoint(self):24 """25 Send two requests to the same path at the same time. This is to ensure we can handle26 multiple requests at once and do not block/queue up requests27 """28 number_of_requests = 1029 start_time = time()30 thread_pool = ThreadPoolExecutor(number_of_requests)31 futures = [32 thread_pool.submit(self.lambda_client.invoke, FunctionName="HelloWorldSleepFunction")33 for _ in range(0, number_of_requests)34 ]35 results = [r.result() for r in as_completed(futures)]36 end_time = time()37 self.assertEqual(len(results), 10)38 self.assertGreater(end_time - start_time, 10)39 for result in results:40 self.assertEqual(result.get("Payload").read().decode("utf-8"), '"Slept for 10s"')41class TestLambdaServiceErrorCases(StartLambdaIntegBaseClass):42 template_path = "/testdata/invoke/template.yml"43 def setUp(self):44 self.url = "http://127.0.0.1:{}".format(self.port)45 self.lambda_client = boto3.client(46 "lambda",47 endpoint_url=self.url,48 region_name="us-east-1",49 use_ssl=False,50 verify=False,51 config=Config(signature_version=UNSIGNED, read_timeout=120, retries={"max_attempts": 0}),52 )53 @pytest.mark.flaky(reruns=3)54 @pytest.mark.timeout(timeout=300, method="thread")55 def test_invoke_with_non_json_data(self):56 expected_error_message = (57 "An error occurred (InvalidRequestContent) when calling the Invoke operation: "58 "Could not parse request body into json: No JSON object could be decoded"59 )60 with self.assertRaises(ClientError) as error:61 self.lambda_client.invoke(FunctionName="EchoEventFunction", Payload="notat:asdfasdf")62 self.assertEqual(str(error.exception), expected_error_message)63 @pytest.mark.flaky(reruns=3)64 @pytest.mark.timeout(timeout=300, method="thread")65 def test_invoke_with_log_type_not_None(self):66 expected_error_message = (67 "An error occurred (NotImplemented) when calling the Invoke operation: "68 "log-type: Tail is not supported. None is only supported."69 )70 with self.assertRaises(ClientError) as error:71 self.lambda_client.invoke(FunctionName="EchoEventFunction", LogType="Tail")72 self.assertEqual(str(error.exception), expected_error_message)73 @pytest.mark.flaky(reruns=3)74 @pytest.mark.timeout(timeout=300, method="thread")75 def test_invoke_with_invocation_type_not_RequestResponse(self):76 expected_error_message = (77 "An error occurred (NotImplemented) when calling the Invoke operation: "78 "invocation-type: DryRun is not supported. RequestResponse is only supported."79 )80 with self.assertRaises(ClientError) as error:81 self.lambda_client.invoke(FunctionName="EchoEventFunction", InvocationType="DryRun")82 self.assertEqual(str(error.exception), expected_error_message)83class TestLambdaService(StartLambdaIntegBaseClass):84 template_path = "/testdata/invoke/template.yml"85 def setUp(self):86 self.url = "http://127.0.0.1:{}".format(self.port)87 self.lambda_client = boto3.client(88 "lambda",89 endpoint_url=self.url,90 region_name="us-east-1",91 use_ssl=False,92 verify=False,93 config=Config(signature_version=UNSIGNED, read_timeout=120, retries={"max_attempts": 0}),94 )95 @pytest.mark.flaky(reruns=3)96 @pytest.mark.timeout(timeout=300, method="thread")97 def test_invoke_with_data(self):98 response = self.lambda_client.invoke(FunctionName="EchoEventFunction", Payload='"This is json data"')99 self.assertEqual(response.get("Payload").read().decode("utf-8"), '"This is json data"')100 self.assertIsNone(response.get("FunctionError"))101 self.assertEqual(response.get("StatusCode"), 200)102 @pytest.mark.flaky(reruns=3)103 @pytest.mark.timeout(timeout=300, method="thread")104 def test_invoke_with_no_data(self):105 response = self.lambda_client.invoke(FunctionName="EchoEventFunction")106 self.assertEqual(response.get("Payload").read().decode("utf-8"), "{}")107 self.assertIsNone(response.get("FunctionError"))108 self.assertEqual(response.get("StatusCode"), 200)109 @pytest.mark.flaky(reruns=3)110 @pytest.mark.timeout(timeout=300, method="thread")111 def test_invoke_with_log_type_None(self):112 response = self.lambda_client.invoke(FunctionName="EchoEventFunction", LogType="None")113 self.assertEqual(response.get("Payload").read().decode("utf-8"), "{}")114 self.assertIsNone(response.get("FunctionError"))115 self.assertEqual(response.get("StatusCode"), 200)116 @pytest.mark.flaky(reruns=3)117 @pytest.mark.timeout(timeout=300, method="thread")118 def test_invoke_with_invocation_type_RequestResponse(self):119 response = self.lambda_client.invoke(FunctionName="EchoEventFunction", InvocationType="RequestResponse")120 self.assertEqual(response.get("Payload").read().decode("utf-8"), "{}")121 self.assertIsNone(response.get("FunctionError"))122 self.assertEqual(response.get("StatusCode"), 200)123 @pytest.mark.flaky(reruns=3)124 @pytest.mark.timeout(timeout=300, method="thread")125 def test_lambda_function_raised_error(self):126 response = self.lambda_client.invoke(FunctionName="RaiseExceptionFunction", InvocationType="RequestResponse")127 self.assertEqual(128 response.get("Payload").read().decode("utf-8"),129 '{"errorMessage": "Lambda is raising an exception", '130 '"errorType": "Exception", '131 '"stackTrace": [["/var/task/main.py", 48, "raise_exception", '132 '"raise Exception(\\"Lambda is raising an exception\\")"]]}',133 )134 self.assertEqual(response.get("FunctionError"), "Unhandled")135 self.assertEqual(response.get("StatusCode"), 200)136 @pytest.mark.flaky(reruns=3)137 @pytest.mark.timeout(timeout=300, method="thread")138 def test_invoke_with_function_timeout(self):139 """140 This behavior does not match the actually Lambda Service. For functions that timeout, data returned like the141 following:142 {"errorMessage":"<timestamp> <request_id> Task timed out after 5.00 seconds"}143 For Local Lambda's, however, timeouts are an interrupt on the thread that runs invokes the function. Since the144 invoke is on a different thread, we do not (currently) have a way to communicate this back to the caller. So145 when a timeout happens locally, we do not add the FunctionError: Unhandled to the response and have an empty146 string as the data returned (because no data was found in stdout from the container).147 """148 response = self.lambda_client.invoke(FunctionName="TimeoutFunction")149 self.assertEqual(response.get("Payload").read().decode("utf-8"), "")150 self.assertIsNone(response.get("FunctionError"))...
test_lambda_basics.py
Source:test_lambda_basics.py
1# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.2# SPDX-License-Identifier: Apache-2.03"""4Unit tests for lambda_basics.py functions.5"""6import json7import unittest.mock8import zipfile9import boto310from botocore.exceptions import ClientError11import pytest12from lambda_basics import LambdaWrapper13def test_create_lambda_deployment_package(monkeypatch):14 monkeypatch.setattr(zipfile.ZipFile, 'write', lambda x, y, z: None)15 wrapper = LambdaWrapper(None, None)16 got_package = wrapper.create_deployment_package('test-file', 'other-file')17 assert got_package is not None18@pytest.mark.parametrize(19 'error_code,stop_on_method', [20 (None, None),21 ('TestException', 'stub_create_role'),22 ('TestException', 'stub_attach_role_policy')23 ])24def test_create_iam_role_for_lambda(25 make_stubber, make_unique_name, stub_runner, error_code, stop_on_method):26 iam_resource = boto3.resource('iam')27 iam_stubber = make_stubber(iam_resource.meta.client)28 wrapper = LambdaWrapper(None, iam_resource)29 role_name = make_unique_name('role-')30 with stub_runner(error_code, stop_on_method) as runner:31 runner.add(iam_stubber.stub_get_role, role_name, error_code='NoSuchEntity')32 runner.add(iam_stubber.stub_create_role, role_name)33 runner.add(34 iam_stubber.stub_attach_role_policy, role_name,35 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole')36 if error_code is None:37 got_role, got_created = wrapper.create_iam_role_for_lambda(role_name)38 assert got_role.name == role_name39 assert got_created40 else:41 with pytest.raises(ClientError) as exc_info:42 wrapper.create_iam_role_for_lambda(role_name)43 assert exc_info.value.response['Error']['Code'] == error_code44@pytest.mark.parametrize('error_code', [None, 'TestException'])45def test_create_function(make_stubber, make_unique_name, error_code):46 lambda_client = boto3.client('lambda')47 lambda_stubber = make_stubber(lambda_client)48 wrapper = LambdaWrapper(lambda_client, None)49 func_name = make_unique_name('func-')50 handler_name = make_unique_name('handler-')51 iam_role = unittest.mock.MagicMock(arn='arn:aws:iam:::role/test-role')52 test_package = 'test-package'53 func_arn = f'arn:aws:lambda:::function/{func_name}'54 lambda_stubber.stub_create_function(55 func_name, func_arn, iam_role.arn, handler_name, test_package,56 error_code=error_code)57 if error_code is None:58 lambda_stubber.stub_get_function(func_name, 'Active')59 if error_code is None:60 got_arn = wrapper.create_function(func_name, handler_name, iam_role, test_package)61 assert got_arn == func_arn62 else:63 with pytest.raises(ClientError) as exc_info:64 wrapper.create_function(65 func_name, handler_name, iam_role, test_package)66 assert exc_info.value.response['Error']['Code'] == error_code67@pytest.mark.parametrize('error_code', [68 None, 'TestException', 'ResourceNotFoundException'])69def test_get_function(make_stubber, error_code):70 lambda_client = boto3.client('lambda')71 lambda_stubber = make_stubber(lambda_client)72 wrapper = LambdaWrapper(lambda_client, None)73 func_name = 'test-func_name'74 lambda_stubber.stub_get_function(func_name, error_code=error_code)75 if error_code in (None, 'ResourceNotFoundException'):76 wrapper.get_function(func_name)77 else:78 with pytest.raises(ClientError) as exc_info:79 wrapper.get_function(func_name)80 assert exc_info.value.response['Error']['Code'] == error_code81@pytest.mark.parametrize('error_code', [None, 'TestException'])82def test_delete_function(make_stubber, make_unique_name, error_code):83 lambda_client = boto3.client('lambda')84 lambda_stubber = make_stubber(lambda_client)85 wrapper = LambdaWrapper(lambda_client, None)86 func_name = make_unique_name('func-')87 lambda_stubber.stub_delete_function(func_name, error_code=error_code)88 if error_code is None:89 wrapper.delete_function(func_name)90 else:91 with pytest.raises(ClientError) as exc_info:92 wrapper.delete_function(func_name)93 assert exc_info.value.response['Error']['Code'] == error_code94@pytest.mark.parametrize('error_code', [None, 'TestException'])95def test_invoke_function(make_stubber, make_unique_name, error_code):96 lambda_client = boto3.client('lambda')97 lambda_stubber = make_stubber(lambda_client)98 wrapper = LambdaWrapper(lambda_client, None)99 func_name = make_unique_name('func-')100 func_params = {'param1': 'test', 'param2': 35}101 response_payload = 'ahoy there'102 lambda_stubber.stub_invoke(103 func_name, json.dumps(func_params), response_payload, log_type='None',104 error_code=error_code)105 if error_code is None:106 response = wrapper.invoke_function(func_name, func_params)107 assert response['Payload'] == response_payload108 else:109 with pytest.raises(ClientError) as exc_info:110 wrapper.invoke_function(func_name, func_params)111 assert exc_info.value.response['Error']['Code'] == error_code112@pytest.mark.parametrize('error_code', [None, 'TestException'])113def test_update_function_code(make_stubber, error_code):114 lambda_client = boto3.client('lambda')115 lambda_stubber = make_stubber(lambda_client)116 wrapper = LambdaWrapper(lambda_client, None)117 func_name = 'test-func_name'118 package = 'test-package'119 update_status = 'InProgress'120 lambda_stubber.stub_update_function_code(121 func_name, update_status, package=package, error_code=error_code)122 if error_code is None:123 got_response = wrapper.update_function_code(func_name, package)124 assert got_response['LastUpdateStatus'] == update_status125 else:126 with pytest.raises(ClientError) as exc_info:127 wrapper.update_function_code(func_name, package)128 assert exc_info.value.response['Error']['Code'] == error_code129@pytest.mark.parametrize('error_code', [None, 'TestException'])130def test_update_function_configuration(make_stubber, error_code):131 lambda_client = boto3.client('lambda')132 lambda_stubber = make_stubber(lambda_client)133 wrapper = LambdaWrapper(lambda_client, None)134 func_name = 'test-func_name'135 env_vars = {'test-key': 'test-val'}136 lambda_stubber.stub_update_function_configuration(137 func_name, env_vars, error_code=error_code)138 if error_code is None:139 got_response = wrapper.update_function_configuration(func_name, env_vars)140 assert got_response141 else:142 with pytest.raises(ClientError) as exc_info:143 wrapper.update_function_configuration(func_name, env_vars)144 assert exc_info.value.response['Error']['Code'] == error_code145@pytest.mark.parametrize('error_code', [None, 'TestException'])146def test_list_functions(make_stubber, error_code):147 lambda_client = boto3.client('lambda')148 lambda_stubber = make_stubber(lambda_client)149 wrapper = LambdaWrapper(lambda_client, None)150 funcs = [{151 'FunctionName': f'test-func-{index}',152 'Description': f'test description {index}',153 'Runtime': f'test-runtime-{index}',154 'Handler': f'test-handler-{index}'}155 for index in range(3)]156 lambda_stubber.stub_list_functions(funcs, error_code=error_code)157 if error_code is None:158 wrapper.list_functions()159 else:160 with pytest.raises(ClientError) as exc_info:161 wrapper.list_functions()...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
