How to use delete_lambda_function method in localstack

Best Python code snippet using localstack_python

test_lambda.py

Source:test_lambda.py

...109            runtime=LAMBDA_RUNTIME_PYTHON27110        )111    @classmethod112    def tearDownClass(cls):113        testutil.delete_lambda_function(TEST_LAMBDA_NAME_PY)114    def test_invocation_type_not_set(self):115        result = self.lambda_client.invoke(116            FunctionName=TEST_LAMBDA_NAME_PY, Payload=b'{}')117        result_data = json.loads(result['Payload'].read())118        self.assertEqual(result['StatusCode'], 200)119        self.assertEqual(result_data['event'], json.loads('{}'))120    def test_invocation_type_request_response(self):121        result = self.lambda_client.invoke(122            FunctionName=TEST_LAMBDA_NAME_PY,123            Payload=b'{}', InvocationType='RequestResponse')124        result_data = result['Payload'].read()125        result_data = json.loads(to_str(result_data))126        self.assertEqual(result['StatusCode'], 200)127        self.assertIsInstance(result_data, dict)128    def test_invocation_type_event(self):129        result = self.lambda_client.invoke(130            FunctionName=TEST_LAMBDA_NAME_PY,131            Payload=b'{}', InvocationType='Event')132        self.assertEqual(result['StatusCode'], 202)133    def test_invocation_type_dry_run(self):134        result = self.lambda_client.invoke(135            FunctionName=TEST_LAMBDA_NAME_PY, Payload=b'{}',136            InvocationType='DryRun')137        self.assertEqual(result['StatusCode'], 204)138    def test_lambda_environment(self):139        zip_file = testutil.create_lambda_archive(140            load_file(TEST_LAMBDA_ENV),141            get_content=True,142            libs=TEST_LAMBDA_LIBS,143            runtime=LAMBDA_RUNTIME_PYTHON27144        )145        testutil.create_lambda_function(146            func_name=TEST_LAMBDA_NAME_ENV,147            zip_file=zip_file,148            runtime=LAMBDA_RUNTIME_PYTHON27,149            envvars={'Hello': 'World'}150        )151        result = self.lambda_client.invoke(152            FunctionName=TEST_LAMBDA_NAME_ENV, Payload=b'{}')153        result_data = result['Payload']154        self.assertEqual(result['StatusCode'], 200)155        self.assertDictEqual(json.load(result_data), {'Hello': 'World'})156        # clean up157        testutil.delete_lambda_function(TEST_LAMBDA_NAME_ENV)158    def test_invocation_with_qualifier(self):159        lambda_name = 'test_lambda_%s' % short_uid()160        bucket_name = 'test_bucket_lambda2'161        bucket_key = 'test_lambda.zip'162        # upload zip file to S3163        zip_file = testutil.create_lambda_archive(164            load_file(TEST_LAMBDA_PYTHON),165            get_content=True,166            libs=TEST_LAMBDA_LIBS,167            runtime=LAMBDA_RUNTIME_PYTHON27168        )169        self.s3_client.create_bucket(Bucket=bucket_name)170        self.s3_client.upload_fileobj(171            BytesIO(zip_file), bucket_name, bucket_key)172        # create lambda function173        response = self.lambda_client.create_function(174            FunctionName=lambda_name, Handler='handler.handler',175            Runtime=lambda_api.LAMBDA_RUNTIME_PYTHON27, Role='r1',176            Code={177                'S3Bucket': bucket_name,178                'S3Key': bucket_key179            },180            Publish=True181        )182        self.assertIn('Version', response)183        # invoke lambda function184        data_before = b'{"foo": "bar with \'quotes\\""}'185        result = self.lambda_client.invoke(186            FunctionName=lambda_name,187            Payload=data_before,188            Qualifier=response['Version']189        )190        data_after = json.loads(result['Payload'].read())191        self.assertEqual(json.loads(to_str(data_before)), data_after['event'])192        context = data_after['context']193        self.assertEqual(response['Version'], context['function_version'])194        self.assertEqual(lambda_name, context['function_name'])195        # assert that logs are present196        expected = ['Lambda log message - print function']197        if use_docker():198            # Note that during regular test execution, nosetests captures the output from199            # the logging module - hence we can only expect this when running in Docker200            expected.append('.*Lambda log message - logging module')201        self.check_lambda_logs(lambda_name, expected_lines=expected)202        # clean up203        testutil.delete_lambda_function(lambda_name)204    def test_upload_lambda_from_s3(self):205        lambda_name = 'test_lambda_%s' % short_uid()206        bucket_name = 'test_bucket_lambda'207        bucket_key = 'test_lambda.zip'208        # upload zip file to S3209        zip_file = testutil.create_lambda_archive(210            load_file(TEST_LAMBDA_PYTHON),211            get_content=True,212            libs=TEST_LAMBDA_LIBS,213            runtime=LAMBDA_RUNTIME_PYTHON27214        )215        self.s3_client.create_bucket(Bucket=bucket_name)216        self.s3_client.upload_fileobj(217            BytesIO(zip_file), bucket_name, bucket_key)218        # create lambda function219        self.lambda_client.create_function(220            FunctionName=lambda_name, Handler='handler.handler',221            Runtime=lambda_api.LAMBDA_RUNTIME_PYTHON27, Role='r1',222            Code={223                'S3Bucket': bucket_name,224                'S3Key': bucket_key225            }226        )227        # invoke lambda function228        data_before = b'{"foo": "bar with \'quotes\\""}'229        result = self.lambda_client.invoke(230            FunctionName=lambda_name, Payload=data_before)231        data_after = json.loads(result['Payload'].read())232        self.assertEqual(json.loads(to_str(data_before)), data_after['event'])233        context = data_after['context']234        self.assertEqual('$LATEST', context['function_version'])235        self.assertEqual(lambda_name, context['function_name'])236        # clean up237        testutil.delete_lambda_function(lambda_name)238    def test_python_lambda_running_in_docker(self):239        if not use_docker():240            return241        zip_file = testutil.create_lambda_archive(242            load_file(TEST_LAMBDA_PYTHON3),243            get_content=True,244            libs=TEST_LAMBDA_LIBS,245            runtime=LAMBDA_RUNTIME_PYTHON36246        )247        testutil.create_lambda_function(248            func_name=TEST_LAMBDA_NAME_PY3,249            zip_file=zip_file,250            runtime=LAMBDA_RUNTIME_PYTHON36251        )252        result = self.lambda_client.invoke(253            FunctionName=TEST_LAMBDA_NAME_PY3, Payload=b'{}')254        result_data = result['Payload'].read()255        self.assertEqual(result['StatusCode'], 200)256        self.assertEqual(to_str(result_data).strip(), '{}')257        # clean up258        testutil.delete_lambda_function(TEST_LAMBDA_NAME_PY3)259class TestNodeJSRuntimes(LambdaTestBase):260    @classmethod261    def setUpClass(cls):262        cls.lambda_client = aws_stack.connect_to_service('lambda')263    def test_nodejs_lambda_running_in_docker(self):264        if not use_docker():265            return266        zip_file = testutil.create_zip_file(267            TEST_LAMBDA_NODEJS, get_content=True)268        testutil.create_lambda_function(269            func_name=TEST_LAMBDA_NAME_JS,270            zip_file=zip_file,271            handler='lambda_integration.handler',272            runtime=LAMBDA_RUNTIME_NODEJS810273        )274        result = self.lambda_client.invoke(275            FunctionName=TEST_LAMBDA_NAME_JS, Payload=b'{}')276        result_data = result['Payload'].read()277        self.assertEqual(result['StatusCode'], 200)278        self.assertEqual(to_str(result_data).strip(), '{}')279        # assert that logs are present280        expected = ['.*Node.js Lambda handler executing.']281        self.check_lambda_logs(TEST_LAMBDA_NAME_JS, expected_lines=expected)282        # clean up283        testutil.delete_lambda_function(TEST_LAMBDA_NAME_JS)284class TestCustomRuntimes(LambdaTestBase):285    @classmethod286    def setUpClass(cls):287        cls.lambda_client = aws_stack.connect_to_service('lambda')288    def test_nodejs_lambda_running_in_docker(self):289        if not use_docker():290            return291        zip_file = testutil.create_zip_file(292            TEST_LAMBDA_CUSTOM_RUNTIME, get_content=True)293        testutil.create_lambda_function(294            func_name=TEST_LAMBDA_NAME_CUSTOM_RUNTIME,295            zip_file=zip_file,296            handler='function.handler',297            runtime=LAMBDA_RUNTIME_CUSTOM_RUNTIME298        )299        result = self.lambda_client.invoke(300            FunctionName=TEST_LAMBDA_NAME_CUSTOM_RUNTIME,301            Payload=b'{"text":"bar with \'quotes\\""}')302        result_data = result['Payload'].read()303        self.assertEqual(result['StatusCode'], 200)304        self.assertEqual(305            to_str(result_data).strip(),306            """Echoing request: '{"text": "bar with \'quotes\\""}'""")307        # assert that logs are present308        expected = ['.*Custom Runtime Lambda handler executing.']309        self.check_lambda_logs(310            TEST_LAMBDA_NAME_CUSTOM_RUNTIME, expected_lines=expected)311        # clean up312        testutil.delete_lambda_function(TEST_LAMBDA_NAME_CUSTOM_RUNTIME)313class TestDotNetCoreRuntimes(LambdaTestBase):314    @classmethod315    def setUpClass(cls):316        cls.lambda_client = aws_stack.connect_to_service('lambda')317        # lambda .NET Core 2.0 is already a zip318        zip_file = TEST_LAMBDA_DOTNETCORE2319        cls.zip_file_content = None320        with open(zip_file, 'rb') as file_obj:321            cls.zip_file_content = file_obj.read()322    def test_dotnet_lambda_running_in_docker(self):323        if not use_docker():324            return325        testutil.create_lambda_function(326            func_name=TEST_LAMBDA_NAME_DOTNETCORE2,327            zip_file=self.zip_file_content,328            handler='DotNetCore2::DotNetCore2.Lambda.Function::SimpleFunctionHandler',329            runtime=LAMBDA_RUNTIME_DOTNETCORE2330        )331        result = self.lambda_client.invoke(332            FunctionName=TEST_LAMBDA_NAME_DOTNETCORE2, Payload=b'{}')333        result_data = result['Payload'].read()334        self.assertEqual(result['StatusCode'], 200)335        self.assertEqual(to_str(result_data).strip(), '{}')336        # assert that logs are present337        expected = ['Running .NET Core 2.0 Lambda']338        self.check_lambda_logs(TEST_LAMBDA_NAME_DOTNETCORE2, expected_lines=expected)339        # clean up340        testutil.delete_lambda_function(TEST_LAMBDA_NAME_DOTNETCORE2)341class TestRubyRuntimes(LambdaTestBase):342    @classmethod343    def setUpClass(cls):344        cls.lambda_client = aws_stack.connect_to_service('lambda')345    def test_ruby_lambda_running_in_docker(self):346        if not use_docker():347            return348        zip_file = testutil.create_zip_file(349            TEST_LAMBDA_RUBY, get_content=True)350        testutil.create_lambda_function(351            func_name=TEST_LAMBDA_NAME_RUBY,352            zip_file=zip_file,353            handler='lambda_integration.handler',354            runtime=LAMBDA_RUNTIME_RUBY25355        )356        result = self.lambda_client.invoke(357            FunctionName=TEST_LAMBDA_NAME_RUBY, Payload=b'{}')358        result_data = result['Payload'].read()359        self.assertEqual(result['StatusCode'], 200)360        self.assertEqual(to_str(result_data).strip(), '{}')361        # clean up362        testutil.delete_lambda_function(TEST_LAMBDA_NAME_RUBY)363class TestJavaRuntimes(LambdaTestBase):364    @classmethod365    def setUpClass(cls):366        cls.lambda_client = aws_stack.connect_to_service('lambda')367        # deploy lambda - Java368        if not os.path.exists(TEST_LAMBDA_JAVA):369            mkdir(os.path.dirname(TEST_LAMBDA_JAVA))370            download(TEST_LAMBDA_JAR_URL, TEST_LAMBDA_JAVA)371        # Lambda supports single JAR deployments without the zip,372        # so we upload the JAR directly.373        cls.test_java_jar = load_file(TEST_LAMBDA_JAVA, mode='rb')374        testutil.create_lambda_function(375            func_name=TEST_LAMBDA_NAME_JAVA,376            zip_file=cls.test_java_jar,377            runtime=LAMBDA_RUNTIME_JAVA8,378            handler='cloud.localstack.sample.LambdaHandler'379        )380        # deploy lambda - Java with stream handler381        testutil.create_lambda_function(382            func_name=TEST_LAMBDA_NAME_JAVA_STREAM,383            zip_file=cls.test_java_jar,384            runtime=LAMBDA_RUNTIME_JAVA8,385            handler='cloud.localstack.sample.LambdaStreamHandler'386        )387        # deploy lambda - Java with serializable input object388        testutil.create_lambda_function(389            func_name=TEST_LAMBDA_NAME_JAVA_SERIALIZABLE,390            zip_file=cls.test_java_jar,391            runtime=LAMBDA_RUNTIME_JAVA8,392            handler='cloud.localstack.sample.SerializedInputLambdaHandler'393        )394    @classmethod395    def tearDownClass(cls):396        # clean up397        testutil.delete_lambda_function(TEST_LAMBDA_NAME_JAVA)398        testutil.delete_lambda_function(TEST_LAMBDA_NAME_JAVA_STREAM)399        testutil.delete_lambda_function(TEST_LAMBDA_NAME_JAVA_SERIALIZABLE)400    def test_java_runtime(self):401        self.assertIsNotNone(self.test_java_jar)402        result = self.lambda_client.invoke(403            FunctionName=TEST_LAMBDA_NAME_JAVA, Payload=b'{}')404        result_data = result['Payload'].read()405        self.assertEqual(result['StatusCode'], 200)406        self.assertIn('LinkedHashMap', to_str(result_data))407    def test_sns_event(self):408        result = self.lambda_client.invoke(409            FunctionName=TEST_LAMBDA_NAME_JAVA, InvocationType='Event',410            Payload=b'{"Records": [{"Sns": {"Message": "{}"}}]}')411        self.assertEqual(result['StatusCode'], 202)412    def test_ddb_event(self):413        result = self.lambda_client.invoke(414            FunctionName=TEST_LAMBDA_NAME_JAVA, InvocationType='Event',415            Payload=b'{"Records": [{"dynamodb": {"Message": "{}"}}]}')416        self.assertEqual(result['StatusCode'], 202)417    def test_kinesis_invocation(self):418        result = self.lambda_client.invoke(419            FunctionName=TEST_LAMBDA_NAME_JAVA,420            Payload=b'{"Records": [{"Kinesis": {"Data": "data", "PartitionKey": "partition"}}]}')421        result_data = result['Payload'].read()422        self.assertEqual(result['StatusCode'], 200)423        self.assertIn('KinesisEvent', to_str(result_data))424    def test_kinesis_event(self):425        result = self.lambda_client.invoke(426            FunctionName=TEST_LAMBDA_NAME_JAVA, InvocationType='Event',427            Payload=b'{"Records": [{"Kinesis": {"Data": "data", "PartitionKey": "partition"}}]}')428        result_data = result['Payload'].read()429        self.assertEqual(result['StatusCode'], 202)430        self.assertEqual(to_str(result_data).strip(), '')431    def test_stream_handler(self):432        result = self.lambda_client.invoke(433            FunctionName=TEST_LAMBDA_NAME_JAVA_STREAM, Payload=b'{}')434        result_data = result['Payload'].read()435        self.assertEqual(result['StatusCode'], 200)436        self.assertEqual(to_str(result_data).strip(), '{}')437    def test_serializable_input_object(self):438        result = self.lambda_client.invoke(439            FunctionName=TEST_LAMBDA_NAME_JAVA_SERIALIZABLE,440            Payload=b'{"bucket": "test_bucket", "key": "test_key"}')441        result_data = result['Payload'].read()442        self.assertEqual(result['StatusCode'], 200)443        self.assertDictEqual(444            json.loads(to_str(result_data)),445            {'validated': True, 'bucket': 'test_bucket', 'key': 'test_key'}446        )447class TestDockerBehaviour(LambdaTestBase):448    @classmethod449    def setUpClass(cls):450        cls.lambda_client = aws_stack.connect_to_service('lambda')451    def test_prime_and_destroy_containers(self):452        # run these tests only for the "reuse containers" Lambda executor453        if not isinstance(lambda_api.LAMBDA_EXECUTOR,454                          lambda_executors.LambdaExecutorReuseContainers):455            return456        executor = lambda_api.LAMBDA_EXECUTOR457        func_name = 'test_prime_and_destroy_containers'458        func_arn = lambda_api.func_arn(func_name)459        # make sure existing containers are gone460        executor.cleanup()461        self.assertEqual(len(executor.get_all_container_names()), 0)462        # deploy and invoke lambda without Docker463        zip_file = testutil.create_lambda_archive(464            load_file(TEST_LAMBDA_ENV),465            get_content=True,466            libs=TEST_LAMBDA_LIBS,467            runtime=LAMBDA_RUNTIME_PYTHON27468        )469        testutil.create_lambda_function(470            func_name=func_name,471            zip_file=zip_file,472            runtime=LAMBDA_RUNTIME_PYTHON27,473            envvars={'Hello': 'World'}474        )475        self.assertEqual(len(executor.get_all_container_names()), 0)476        self.assertDictEqual(executor.function_invoke_times, {})477        # invoke a few times.478        durations = []479        num_iterations = 3480        for i in range(0, num_iterations + 1):481            prev_invoke_time = None482            if i > 0:483                prev_invoke_time = executor.function_invoke_times[func_arn]484            start_time = time.time()485            self.lambda_client.invoke(FunctionName=func_name, Payload=b'{}')486            duration = time.time() - start_time487            self.assertEqual(len(executor.get_all_container_names()), 1)488            # ensure the last invoke time is being updated properly.489            if i > 0:490                self.assertGreater(executor.function_invoke_times[func_arn], prev_invoke_time)491            else:492                self.assertGreater(executor.function_invoke_times[func_arn], 0)493            durations.append(duration)494        # the first call would have created the container. subsequent calls would reuse and be faster.495        for i in range(1, num_iterations + 1):496            self.assertLess(durations[i], durations[0])497        status = executor.get_docker_container_status(func_arn)498        self.assertEqual(status, 1)499        container_network = executor.get_docker_container_network(func_arn)500        self.assertEqual(container_network, 'default')501        executor.cleanup()502        status = executor.get_docker_container_status(func_arn)503        self.assertEqual(status, 0)504        self.assertEqual(len(executor.get_all_container_names()), 0)505        # clean up506        testutil.delete_lambda_function(func_name)507    def test_docker_command_for_separate_container_lambda_executor(self):508        # run these tests only for the "separate containers" Lambda executor509        if not isinstance(lambda_api.LAMBDA_EXECUTOR,510                          lambda_executors.LambdaExecutorSeparateContainers):511            return512        executor = lambda_api.LAMBDA_EXECUTOR513        func_name = 'test_docker_command_for_separate_container_lambda_executor'514        func_arn = lambda_api.func_arn(func_name)515        handler = 'handler'516        lambda_cwd = '/app/lambda'517        network = 'compose_network'518        config.LAMBDA_DOCKER_NETWORK = network519        cmd = executor.prepare_execution(func_arn, {}, LAMBDA_RUNTIME_NODEJS810, '', handler, lambda_cwd)520        expected = 'docker run -v "%s":/var/task   --network="%s"  --rm "lambci/lambda:%s" "%s"' % (521            lambda_cwd, network, LAMBDA_RUNTIME_NODEJS810, handler)522        self.assertIn(('--network="%s"' % network), cmd, 'cmd=%s expected=%s' % (cmd, expected))523        config.LAMBDA_DOCKER_NETWORK = ''524    def test_destroy_idle_containers(self):525        # run these tests only for the "reuse containers" Lambda executor526        if not isinstance(lambda_api.LAMBDA_EXECUTOR,527                          lambda_executors.LambdaExecutorReuseContainers):528            return529        executor = lambda_api.LAMBDA_EXECUTOR530        func_name = 'test_destroy_idle_containers'531        func_arn = lambda_api.func_arn(func_name)532        # make sure existing containers are gone533        executor.destroy_existing_docker_containers()534        self.assertEqual(len(executor.get_all_container_names()), 0)535        # deploy and invoke lambda without Docker536        zip_file = testutil.create_lambda_archive(537            load_file(TEST_LAMBDA_ENV),538            get_content=True,539            libs=TEST_LAMBDA_LIBS,540            runtime=LAMBDA_RUNTIME_PYTHON27541        )542        testutil.create_lambda_function(543            func_name=func_name,544            zip_file=zip_file,545            runtime=LAMBDA_RUNTIME_PYTHON27,546            envvars={'Hello': 'World'}547        )548        self.assertEqual(len(executor.get_all_container_names()), 0)549        self.lambda_client.invoke(FunctionName=func_name, Payload=b'{}')550        self.assertEqual(len(executor.get_all_container_names()), 1)551        # try to destroy idle containers.552        executor.idle_container_destroyer()553        self.assertEqual(len(executor.get_all_container_names()), 1)554        # simulate an idle container555        executor.function_invoke_times[func_arn] = time.time() - lambda_executors.MAX_CONTAINER_IDLE_TIME_MS556        executor.idle_container_destroyer()557        self.assertEqual(len(executor.get_all_container_names()), 0)558        # clean up...

aws_lambda_handler.py

Source:aws_lambda_handler.py

...6    def create_aws_lambda(self):7        _lambda_client = Utils.get_client(self, "lambda", self._lambda_endpoint_url, self._aws_region)8        _exist_result = Utils.exist_lambda_function(self, _lambda_client)9        if _exist_result:10            LambdaHandler.delete_lambda_function(self, _lambda_client)11        LambdaHandler.create_lambda_function(self, _lambda_client)12        # register stream consumer13        _rtn_consumer_map = LambdaHandler.connect_datastream_to_lambda(self, _lambda_client)14        print("[Log] consumer status {} Ok: ".format(_rtn_consumer_map.get("ResponseMetadata").get("HTTPStatusCode")))15    def create_lambda_function(self, _lambda_client):16        LambdaHandler.create_lambda_zip(self)17        with open(self._lambda_zip_path, 'rb') as f:18            zipped_code = f.read()19        _lambda_client.create_function(20            FunctionName=self._lambda_func_name,21            Runtime='python3.6',22            Role='arn:aws:iam::000000000000:role/adminRole',23            Handler=self._lambda_func_name + '.handler',24            Code=dict(ZipFile=zipped_code)25        )26    def create_lambda_zip(self):27        with ZipFile(self._lambda_zip_path, 'w') as z:28            z.write(self._copy_destination_path)29            z.close()30    def invoke_function_and_get_message(self, _lambda_client, invo_type):31        _response = _lambda_client.invoke(32                        FunctionName=self._lambda_func_name,33                        InvocationType=invo_type34                    )35        if invo_type == 'RequestResponse':36            _rtn_data=json.loads(_response['Payload'].read().decode('utf-8'))37        else:38            _rtn_data=_response['StatusCode']39        return _rtn_data40    def delete_lambda_function(self, _lambda_client):41        _response = _lambda_client.delete_function(42                        FunctionName=self._lambda_func_name43                    )44        return _response45    def connect_datastream_to_lambda(self, _lambda_client):46        _res_evt_map = LambdaHandler.create_event_mapping(self, _lambda_client)47        return _res_evt_map48    def create_event_mapping(self, _lambda_client):49        _res_evt_map = _lambda_client.create_event_source_mapping(50                            EventSourceArn='arn:aws:kinesis:us-east-1:000000000000:stream/weblog-kinesis-datastream',51                            FunctionName=self._lambda_func_name,52                            BatchSize=500,53                            StartingPosition='AT_TIMESTAMP',54                            StartingPositionTimestamp=1541139109,...

app.py

Source:app.py

...56            lambda_client, lambda_function_name, lambda_parms)57        print(f"The {lambda_parms['action']} of {lambda_parms['number']} resulted in "58              f"{json.load(response['Payload'])}")59    ### TO USE DELETE LAMBDA FUNCTION ###60    delete_lambda_function(lambda_client, lambda_function_name)61    print(f"Deleted function {lambda_function_name}.")62if __name__ == '__main__':...

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.