Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use lambda_api method in localstack

Best Python code snippet using localstack_python

34132_test_lambda.py

Source:34132_test_lambda.py

1import unittest2import json3import flask4from localstack.services.awslambda import lambda_api5from localstack.utils.aws.aws_models import LambdaFunction6class TestLambdaAPI(unittest.TestCase):7    CODE_SIZE = 508    HANDLER = 'index.handler'9    RUNTIME = 'node.js4.3'10    TIMEOUT = 60  # Default value, hardcoded11    FUNCTION_NAME = 'test1'12    ALIAS_NAME = 'alias1'13    ALIAS2_NAME = 'alias2'14    RESOURCENOTFOUND_EXCEPTION = 'ResourceNotFoundException'15    RESOURCENOTFOUND_MESSAGE = 'Function not found: %s'16    ALIASEXISTS_EXCEPTION = 'ResourceConflictException'17    ALIASEXISTS_MESSAGE = 'Alias already exists: %s'18    ALIASNOTFOUND_EXCEPTION = 'ResourceNotFoundException'19    ALIASNOTFOUND_MESSAGE = 'Alias not found: %s'20    TEST_UUID = 'Test'21    def setUp(self):22        lambda_api.cleanup()23        self.maxDiff = None24        self.app = flask.Flask(__name__)25    def test_delete_event_source_mapping(self):26        with self.app.test_request_context():27            lambda_api.event_source_mappings.append({'UUID': self.TEST_UUID})28            result = lambda_api.delete_event_source_mapping(self.TEST_UUID)29            self.assertEqual(json.loads(result.get_data()).get('UUID'), self.TEST_UUID)30            self.assertEqual(0, len(lambda_api.event_source_mappings))31    def test_publish_function_version(self):32        with self.app.test_request_context():33            self._create_function(self.FUNCTION_NAME)34            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())35            result2 = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())36            expected_result = dict()37            expected_result['CodeSize'] = self.CODE_SIZE38            expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'39            expected_result['FunctionName'] = str(self.FUNCTION_NAME)40            expected_result['Handler'] = str(self.HANDLER)41            expected_result['Runtime'] = str(self.RUNTIME)42            expected_result['Timeout'] = self.TIMEOUT43            expected_result['Version'] = '1'44            expected_result2 = dict(expected_result)45            expected_result2['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'46            expected_result2['Version'] = '2'47            self.assertDictEqual(expected_result, result)48            self.assertDictEqual(expected_result2, result2)49    def test_publish_non_existant_function_version_returns_error(self):50        with self.app.test_request_context():51            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())52            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])53            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),54                             result['message'])55    def test_list_function_versions(self):56        with self.app.test_request_context():57            self._create_function(self.FUNCTION_NAME)58            lambda_api.publish_version(self.FUNCTION_NAME)59            lambda_api.publish_version(self.FUNCTION_NAME)60            result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())61            latest_version = dict()62            latest_version['CodeSize'] = self.CODE_SIZE63            latest_version['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':$LATEST'64            latest_version['FunctionName'] = str(self.FUNCTION_NAME)65            latest_version['Handler'] = str(self.HANDLER)66            latest_version['Runtime'] = str(self.RUNTIME)67            latest_version['Timeout'] = self.TIMEOUT68            latest_version['Version'] = '$LATEST'69            version1 = dict(latest_version)70            version1['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'71            version1['Version'] = '1'72            version2 = dict(latest_version)73            version2['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'74            version2['Version'] = '2'75            expected_result = {'Versions': sorted([latest_version, version1, version2],76                                                  key=lambda k: str(k.get('Version')))}77            self.assertDictEqual(expected_result, result)78    def test_list_non_existant_function_versions_returns_error(self):79        with self.app.test_request_context():80            result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())81            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])82            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),83                             result['message'])84    def test_create_alias(self):85        with self.app.test_request_context():86            self._create_function(self.FUNCTION_NAME)87            lambda_api.publish_version(self.FUNCTION_NAME)88            flask.request.data = json.dumps({89                'Name': self.ALIAS_NAME,90                'FunctionVersion': '1',91                'Description': ''92            })93            result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())94            expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,95                               'FunctionVersion': '1', 'Description': '', 'Name': self.ALIAS_NAME}96            self.assertDictEqual(expected_result, result)97    def test_create_alias_on_non_existant_function_returns_error(self):98        with self.app.test_request_context():99            result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())100            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])101            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),102                             result['message'])103    def test_create_alias_returns_error_if_already_exists(self):104        with self.app.test_request_context():105            self._create_function(self.FUNCTION_NAME)106            lambda_api.publish_version(self.FUNCTION_NAME)107            flask.request.data = json.dumps({108                'Name': self.ALIAS_NAME,109                'FunctionVersion': '1',110                'Description': ''111            })112            lambda_api.create_alias(self.FUNCTION_NAME).get_data()113            result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())114            alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME115            self.assertEqual(self.ALIASEXISTS_EXCEPTION, result['__type'])116            self.assertEqual(self.ALIASEXISTS_MESSAGE % alias_arn,117                             result['message'])118    def test_update_alias(self):119        with self.app.test_request_context():120            self._create_function(self.FUNCTION_NAME)121            lambda_api.publish_version(self.FUNCTION_NAME)122            flask.request.data = json.dumps({123                'Name': self.ALIAS_NAME,124                'FunctionVersion': '1',125                'Description': ''126            })127            lambda_api.create_alias(self.FUNCTION_NAME).get_data()128            flask.request.data = json.dumps({129                'FunctionVersion': '$LATEST',130                'Description': 'Test-Description'131            })132            result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())133            expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,134                               'FunctionVersion': '$LATEST', 'Description': 'Test-Description',135                               'Name': self.ALIAS_NAME}136            self.assertDictEqual(expected_result, result)137    def test_update_alias_on_non_existant_function_returns_error(self):138        with self.app.test_request_context():139            result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())140            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])141            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),142                             result['message'])143    def test_update_alias_on_non_existant_alias_returns_error(self):144        with self.app.test_request_context():145            self._create_function(self.FUNCTION_NAME)146            result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())147            alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME148            self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])149            self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])150    def test_list_aliases(self):151        with self.app.test_request_context():152            self._create_function(self.FUNCTION_NAME)153            lambda_api.publish_version(self.FUNCTION_NAME)154            flask.request.data = json.dumps({155                'Name': self.ALIAS2_NAME,156                'FunctionVersion': '$LATEST'157            })158            lambda_api.create_alias(self.FUNCTION_NAME).get_data()159            flask.request.data = json.dumps({160                'Name': self.ALIAS_NAME,161                'FunctionVersion': '1',162                'Description': self.ALIAS_NAME163            })164            lambda_api.create_alias(self.FUNCTION_NAME).get_data()165            result = json.loads(lambda_api.list_aliases(self.FUNCTION_NAME).get_data())166            expected_result = {'Aliases': [167                {168                    'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,169                    'FunctionVersion': '1',170                    'Name': self.ALIAS_NAME,171                    'Description': self.ALIAS_NAME172                },173                {174                    'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS2_NAME,175                    'FunctionVersion': '$LATEST',176                    'Name': self.ALIAS2_NAME,177                    'Description': ''178                }179            ]}180            self.assertDictEqual(expected_result, result)181    def test_list_non_existant_function_aliases_returns_error(self):182        with self.app.test_request_context():183            result = json.loads(lambda_api.list_aliases(self.FUNCTION_NAME).get_data())184            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])185            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),186                             result['message'])187    def _create_function(self, function_name):188        arn = lambda_api.func_arn(function_name)189        lambda_api.arn_to_lambda[arn] = LambdaFunction(arn)190        lambda_api.arn_to_lambda[arn].versions = {'$LATEST': {'CodeSize': self.CODE_SIZE}}191        lambda_api.arn_to_lambda[arn].handler = self.HANDLER192        lambda_api.arn_to_lambda[arn].runtime = self.RUNTIME...

test_lambda.py

Source:test_lambda.py

1import unittest2import json3from flask import Flask4from localstack.services.awslambda import lambda_api5from localstack.utils.aws.aws_models import LambdaFunction6class TestLambdaAPI(unittest.TestCase):7    CODE_SIZE = 508    HANDLER = 'index.handler'9    RUNTIME = 'node.js4.3'10    TIMEOUT = 60  # Default value, hardcoded11    FUNCTION_NAME = 'test1'12    RESOURCENOTFOUND_EXCEPTION = 'ResourceNotFoundException'13    RESOURCENOTFOUND_MESSAGE = 'Function not found: %s'14    TEST_UUID = "Test"15    def setUp(self):16        lambda_api.cleanup()17        self.maxDiff = None18        self.app = Flask(__name__)19    def test_delete_event_source_mapping(self):20        with self.app.test_request_context():21            lambda_api.event_source_mappings.append({'UUID': self.TEST_UUID})22            result = lambda_api.delete_event_source_mapping(self.TEST_UUID)23            self.assertEqual(json.loads(result.get_data()).get('UUID'), self.TEST_UUID)24            self.assertEqual(0, len(lambda_api.event_source_mappings))25    def test_publish_function_version(self):26        with self.app.test_request_context():27            self._create_function(self.FUNCTION_NAME)28            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())29            result2 = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())30            expected_result = dict()31            expected_result[u'CodeSize'] = self.CODE_SIZE32            expected_result[u'FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME))33            expected_result[u'FunctionName'] = str(self.FUNCTION_NAME)34            expected_result[u'Handler'] = str(self.HANDLER)35            expected_result[u'Runtime'] = str(self.RUNTIME)36            expected_result[u'Timeout'] = self.TIMEOUT37            expected_result[u'Version'] = 138            expected_result2 = dict(expected_result)39            expected_result2[u'Version'] = 240            self.assertDictEqual(expected_result, result)41            self.assertDictEqual(expected_result2, result2)42    def test_publish_non_existant_function_version_returns_error(self):43        with self.app.test_request_context():44            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())45            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])46            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),47                             result['message'])48    def test_list_function_versions(self):49        with self.app.test_request_context():50            self._create_function(self.FUNCTION_NAME)51            lambda_api.publish_version(self.FUNCTION_NAME)52            lambda_api.publish_version(self.FUNCTION_NAME)53            result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())54            latest_version = dict()55            latest_version[u'CodeSize'] = self.CODE_SIZE56            latest_version[u'FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME))57            latest_version[u'FunctionName'] = str(self.FUNCTION_NAME)58            latest_version[u'Handler'] = str(self.HANDLER)59            latest_version[u'Runtime'] = str(self.RUNTIME)60            latest_version[u'Timeout'] = self.TIMEOUT61            latest_version[u'Version'] = u'$LATEST'62            version1 = dict(latest_version)63            version1[u'Version'] = 164            version2 = dict(latest_version)65            version2[u'Version'] = 266            expected_result = {u'Versions': sorted([latest_version, version1, version2],67                                                   key=lambda k: str(k.get('Version')))}68            self.assertDictEqual(expected_result, result)69    def test_list_non_existant_function_versions_returns_error(self):70        with self.app.test_request_context():71            result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())72            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])73            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),74                             result['message'])75    def _create_function(self, function_name):76        arn = lambda_api.func_arn(function_name)77        lambda_api.arn_to_lambda[arn] = LambdaFunction(arn)78        lambda_api.arn_to_lambda[arn].versions = {'$LATEST': {'CodeSize': self.CODE_SIZE}}79        lambda_api.arn_to_lambda[arn].handler = self.HANDLER80        lambda_api.arn_to_lambda[arn].runtime = self.RUNTIME...

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.