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

How to use _create_function method in localstack

Best Python code snippet using localstack_python

test_lambda.py

Source:test_lambda.py

...42            self.assertEqual(json.loads(result.get_data()).get('UUID'), self.TEST_UUID)43            self.assertEqual(0, len(lambda_api.event_source_mappings))44    def test_publish_function_version(self):45        with self.app.test_request_context():46            self._create_function(self.FUNCTION_NAME)47            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())48            result2 = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())49            expected_result = dict()50            expected_result['CodeSize'] = self.CODE_SIZE51            expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'52            expected_result['FunctionName'] = str(self.FUNCTION_NAME)53            expected_result['Handler'] = str(self.HANDLER)54            expected_result['Runtime'] = str(self.RUNTIME)55            expected_result['Timeout'] = self.TIMEOUT56            expected_result['Version'] = '1'57            expected_result['Environment'] = {'Variables': {}}58            expected_result2 = dict(expected_result)59            expected_result2['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'60            expected_result2['Version'] = '2'61            expected_result2['Environment'] = {'Variables': {}}62            self.assertDictEqual(expected_result, result)63            self.assertDictEqual(expected_result2, result2)64    def test_publish_non_existant_function_version_returns_error(self):65        with self.app.test_request_context():66            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())67            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])68            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),69                             result['message'])70    def test_list_function_versions(self):71        with self.app.test_request_context():72            self._create_function(self.FUNCTION_NAME)73            lambda_api.publish_version(self.FUNCTION_NAME)74            lambda_api.publish_version(self.FUNCTION_NAME)75            result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())76            latest_version = dict()77            latest_version['CodeSize'] = self.CODE_SIZE78            latest_version['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':$LATEST'79            latest_version['FunctionName'] = str(self.FUNCTION_NAME)80            latest_version['Handler'] = str(self.HANDLER)81            latest_version['Runtime'] = str(self.RUNTIME)82            latest_version['Timeout'] = self.TIMEOUT83            latest_version['Version'] = '$LATEST'84            latest_version['Environment'] = {'Variables': {}}85            version1 = dict(latest_version)86            version1['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'87            version1['Version'] = '1'88            version1['Environment'] = {'Variables': {}}89            version2 = dict(latest_version)90            version2['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'91            version2['Version'] = '2'92            version2['Environment'] = {'Variables': {}}93            expected_result = {'Versions': sorted([latest_version, version1, version2],94                                                  key=lambda k: str(k.get('Version')))}95            self.assertDictEqual(expected_result, result)96    def test_list_non_existant_function_versions_returns_error(self):97        with self.app.test_request_context():98            result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())99            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])100            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),101                             result['message'])102    def test_create_alias(self):103        self._create_function(self.FUNCTION_NAME)104        self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))105        response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),106                         data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',107                             'Description': ''}))108        result = json.loads(response.get_data())109        expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,110                           'FunctionVersion': '1', 'Description': '', 'Name': self.ALIAS_NAME}111        self.assertDictEqual(expected_result, result)112    def test_create_alias_on_non_existant_function_returns_error(self):113        with self.app.test_request_context():114            result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())115            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])116            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),117                             result['message'])118    def test_create_alias_returns_error_if_already_exists(self):119        self._create_function(self.FUNCTION_NAME)120        self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))121        data = json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''})122        self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME), data=data)123        response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),124                                    data=data)125        result = json.loads(response.get_data())126        alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME127        self.assertEqual(self.ALIASEXISTS_EXCEPTION, result['__type'])128        self.assertEqual(self.ALIASEXISTS_MESSAGE % alias_arn,129                         result['message'])130    def test_update_alias(self):131        self._create_function(self.FUNCTION_NAME)132        self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))133        self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),134                         data=json.dumps({135                             'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''}))136        response = self.client.put('{0}/functions/{1}/aliases/{2}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME,137                                                                          self.ALIAS_NAME),138                                   data=json.dumps({'FunctionVersion': '$LATEST', 'Description': 'Test-Description'}))139        result = json.loads(response.get_data())140        expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,141                           'FunctionVersion': '$LATEST', 'Description': 'Test-Description',142                           'Name': self.ALIAS_NAME}143        self.assertDictEqual(expected_result, result)144    def test_update_alias_on_non_existant_function_returns_error(self):145        with self.app.test_request_context():146            result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())147            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])148            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),149                             result['message'])150    def test_update_alias_on_non_existant_alias_returns_error(self):151        with self.app.test_request_context():152            self._create_function(self.FUNCTION_NAME)153            result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())154            alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME155            self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])156            self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])157    def test_get_alias(self):158        self._create_function(self.FUNCTION_NAME)159        self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))160        self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),161                         data=json.dumps({162                             'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''}))163        response = self.client.get('{0}/functions/{1}/aliases/{2}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME,164                                                                          self.ALIAS_NAME))165        result = json.loads(response.get_data())166        expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,167                           'FunctionVersion': '1', 'Description': '',168                           'Name': self.ALIAS_NAME}169        self.assertDictEqual(expected_result, result)170    def test_get_alias_on_non_existant_function_returns_error(self):171        with self.app.test_request_context():172            result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())173            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])174            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),175                             result['message'])176    def test_get_alias_on_non_existant_alias_returns_error(self):177        with self.app.test_request_context():178            self._create_function(self.FUNCTION_NAME)179            result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())180            alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME181            self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])182            self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])183    def test_list_aliases(self):184        self._create_function(self.FUNCTION_NAME)185        self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))186        self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),187                         data=json.dumps({'Name': self.ALIAS2_NAME, 'FunctionVersion': '$LATEST'}))188        self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),189                         data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',190                                          'Description': self.ALIAS_NAME}))191        response = self.client.get('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))192        result = json.loads(response.get_data())193        expected_result = {'Aliases': [194            {195                'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,196                'FunctionVersion': '1',197                'Name': self.ALIAS_NAME,198                'Description': self.ALIAS_NAME199            },200            {201                'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS2_NAME,202                'FunctionVersion': '$LATEST',203                'Name': self.ALIAS2_NAME,204                'Description': ''205            }206        ]}207        self.assertDictEqual(expected_result, result)208    def test_list_non_existant_function_aliases_returns_error(self):209        with self.app.test_request_context():210            result = json.loads(lambda_api.list_aliases(self.FUNCTION_NAME).get_data())211            self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])212            self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),213                             result['message'])214    def test_get_container_name(self):215        executor = lambda_executors.EXECUTOR_CONTAINERS_REUSE216        name = executor.get_container_name('arn:aws:lambda:us-east-1:00000000:function:my_function_name')217        self.assertEqual(name, 'localstack_lambda_arn_aws_lambda_us-east-1_00000000_function_my_function_name')218    def test_put_concurrency(self):219        with self.app.test_request_context():220            self._create_function(self.FUNCTION_NAME)221            # note: PutFunctionConcurrency is mounted at: /2017-10-31222            # NOT lambda_api.PATH_ROOT223            # https://docs.aws.amazon.com/lambda/latest/dg/API_PutFunctionConcurrency.html224            concurrency_data = {'ReservedConcurrentExecutions': 10}225            response = self.client.put('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME),226                                       data=json.dumps(concurrency_data))227            result = json.loads(response.get_data())228            self.assertDictEqual(concurrency_data, result)229    def test_concurrency_get_function(self):230        with self.app.test_request_context():231            self._create_function(self.FUNCTION_NAME)232            # note: PutFunctionConcurrency is mounted at: /2017-10-31233            # NOT lambda_api.PATH_ROOT234            # https://docs.aws.amazon.com/lambda/latest/dg/API_PutFunctionConcurrency.html235            concurrency_data = {'ReservedConcurrentExecutions': 10}236            self.client.put('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME),237                            data=json.dumps(concurrency_data))238            response = self.client.get('{0}/functions/{1}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))239            result = json.loads(response.get_data())240            self.assertTrue('Concurrency' in result)241            self.assertDictEqual(concurrency_data, result['Concurrency'])242    def _create_function(self, function_name):243        arn = lambda_api.func_arn(function_name)244        lambda_api.arn_to_lambda[arn] = LambdaFunction(arn)245        lambda_api.arn_to_lambda[arn].versions = {'$LATEST': {'CodeSize': self.CODE_SIZE}}246        lambda_api.arn_to_lambda[arn].handler = self.HANDLER247        lambda_api.arn_to_lambda[arn].runtime = self.RUNTIME248        lambda_api.arn_to_lambda[arn].timeout = self.TIMEOUT...

test_non_standard_injection.py

Source:test_non_standard_injection.py

...9        self.assertIsInstance(injector.get("foo"), TestClass)10    def test_function_injection(self):11        injector = autoinject.InjectionManager(False)12        @injector.register("foo")13        def _create_function():14            def foo_function(bar):15                return "!{}!".format(bar)16            return foo_function17        f = injector.get("foo")18        self.assertTrue(callable(f))19        self.assertEqual(f("bar"), "!bar!")20    def test_function_override_injection(self):21        injector = autoinject.InjectionManager(False)22        @injector.register("foo")23        def _create_function():24            def foo_function(bar):25                return "!{}!".format(bar)26            return foo_function27        @injector.register("foo")28        def _create_function():29            def foo_function(bar):30                return bar.upper()31            return foo_function32        f = injector.get("foo")33        self.assertTrue(callable(f))34        self.assertNotEqual(f("bar"), "!bar!")35        self.assertEqual(f("bar"), "BAR")36    def test_function_override_with_weight_injection(self):37        injector = autoinject.InjectionManager(False)38        @injector.register("foo", weight=5)39        def _create_function():40            def foo_function(bar):41                return "!{}!".format(bar)42            return foo_function43        @injector.register("foo")44        def _create_function():45            def foo_function(bar):46                return bar.upper()47            return foo_function48        f = injector.get("foo")49        self.assertTrue(callable(f))50        self.assertNotEqual(f("bar"), "BAR")...

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.