How to use create_lambda_function method in localstack

Best Python code snippet using localstack_python

test_lambda_whitebox.py

Source:test_lambda_whitebox.py

...83            self.assertIn("non-existing-lambda", record["headers"]["lambda-function-name"])84        self.assertEqual(3, items_after)85        # create test Lambda86        lambda_name = "test-%s" % short_uid()87        testutil.create_lambda_function(88            handler_file=TEST_LAMBDA_PYTHON,89            func_name=lambda_name,90            libs=TEST_LAMBDA_LIBS,91        )92        # test 2: forward to LAMBDA_FORWARD_URL93        records = []94        inv_results = self._run_forward_to_fallback_url(95            local_url, lambda_name=lambda_name, fallback=False96        )97        items_after = len(records)98        for record in records:99            headers = record["headers"]100            self.assertIn("/lambda/", headers["Authorization"])101            self.assertEqual("POST", record["method"])102            self.assertIn("/functions/%s/invocations" % lambda_name, record["path"])103            self.assertTrue(headers.get("X-Amz-Client-Context"))104            self.assertEqual("RequestResponse", headers.get("X-Amz-Invocation-Type"))105            self.assertEqual({"foo": "bar"}, json.loads(to_str(record["data"])))106        self.assertEqual(3, items_after)107        # assert result payload matches108        response_payload = inv_results[0]["Payload"].read()109        self.assertEqual(lambda_result, json.loads(response_payload))110        # clean up / shutdown111        lambda_client = aws_stack.create_external_boto_client("lambda")112        lambda_client.delete_function(FunctionName=lambda_name)113        proxy.stop()114    def test_adding_fallback_function_name_in_headers(self):115        lambda_client = aws_stack.create_external_boto_client("lambda")116        ddb_client = aws_stack.create_external_boto_client("dynamodb")117        db_table = "lambda-records"118        config.LAMBDA_FALLBACK_URL = "dynamodb://%s" % db_table119        lambda_client.invoke(120            FunctionName="non-existing-lambda",121            Payload=b"{}",122            InvocationType="RequestResponse",123        )124        result = run_safe(ddb_client.scan, TableName=db_table)125        self.assertEqual("non-existing-lambda", result["Items"][0]["function_name"]["S"])126class TestDockerExecutors(unittest.TestCase):127    @classmethod128    def setUpClass(cls):129        cls.lambda_client = aws_stack.create_external_boto_client("lambda")130        cls.s3_client = aws_stack.create_external_boto_client("s3")131    @pytest.mark.skipif(not use_docker(), reason="Only applicable with docker executor")132    def test_additional_docker_flags(self):133        flags_before = config.LAMBDA_DOCKER_FLAGS134        env_value = short_uid()135        config.LAMBDA_DOCKER_FLAGS = f"-e Hello={env_value}"136        function_name = "flags-{}".format(short_uid())137        try:138            testutil.create_lambda_function(139                handler_file=TEST_LAMBDA_ENV,140                libs=TEST_LAMBDA_LIBS,141                func_name=function_name,142            )143            lambda_client = aws_stack.create_external_boto_client("lambda")144            result = lambda_client.invoke(FunctionName=function_name, Payload="{}")145            self.assertEqual(200, result["ResponseMetadata"]["HTTPStatusCode"])146            result_data = result["Payload"].read()147            result_data = json.loads(to_str(result_data))148            self.assertEqual({"Hello": env_value}, result_data)149        finally:150            config.LAMBDA_DOCKER_FLAGS = flags_before151        # clean up152        lambda_client.delete_function(FunctionName=function_name)153    def test_code_updated_on_redeployment(self):154        lambda_api.LAMBDA_EXECUTOR.cleanup()155        func_name = "test_code_updated_on_redeployment"156        # deploy function for the first time157        testutil.create_lambda_function(158            func_name=func_name,159            handler_file=TEST_LAMBDA_ENV,160            libs=TEST_LAMBDA_LIBS,161            envvars={"Hello": "World"},162        )163        # test first invocation164        result = self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")165        payload = json.loads(to_str(result["Payload"].read()))166        assert payload["Hello"] == "World"167        # replacement code168        updated_handler = "handler = lambda event, context: {'Hello': 'Elon Musk'}"169        updated_handler = testutil.create_lambda_archive(170            updated_handler, libs=TEST_LAMBDA_LIBS, get_content=True171        )172        self.lambda_client.update_function_code(FunctionName=func_name, ZipFile=updated_handler)173        # second invocation should exec updated lambda code174        result = self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")175        payload = json.loads(to_str(result["Payload"].read()))176        assert payload["Hello"] == "Elon Musk"177    @pytest.mark.skipif(178        condition=not isinstance(179            lambda_api.LAMBDA_EXECUTOR, lambda_executors.LambdaExecutorReuseContainers180        ),181        reason="Test only applicable if docker-reuse executor is selected",182    )183    def test_prime_and_destroy_containers(self):184        executor = lambda_api.LAMBDA_EXECUTOR185        func_name = "test_prime_and_destroy_containers"186        func_arn = lambda_api.func_arn(func_name)187        # make sure existing containers are gone188        executor.cleanup()189        self.assertEqual(0, len(executor.get_all_container_names()))190        # deploy and invoke lambda without Docker191        testutil.create_lambda_function(192            func_name=func_name,193            handler_file=TEST_LAMBDA_ENV,194            libs=TEST_LAMBDA_LIBS,195            envvars={"Hello": "World"},196        )197        self.assertEqual(0, len(executor.get_all_container_names()))198        self.assertDictEqual({}, executor.function_invoke_times)199        # invoke a few times.200        durations = []201        num_iterations = 3202        for i in range(0, num_iterations + 1):203            prev_invoke_time = None204            if i > 0:205                prev_invoke_time = executor.function_invoke_times[func_arn]206            start_time = time.time()207            self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")208            duration = time.time() - start_time209            self.assertEqual(1, len(executor.get_all_container_names()))210            # ensure the last invoke time is being updated properly.211            if i > 0:212                self.assertGreater(executor.function_invoke_times[func_arn], prev_invoke_time)213            else:214                self.assertGreater(executor.function_invoke_times[func_arn], 0)215            durations.append(duration)216        # the first call would have created the container. subsequent calls would reuse and be faster.217        for i in range(1, num_iterations + 1):218            self.assertLess(durations[i], durations[0])219        status = executor.get_docker_container_status(func_arn)220        self.assertEqual(1, status)221        container_network = executor.get_docker_container_network(func_arn)222        self.assertEqual("bridge", container_network)223        executor.cleanup()224        status = executor.get_docker_container_status(func_arn)225        self.assertEqual(0, status)226        self.assertEqual(0, len(executor.get_all_container_names()))227        # clean up228        testutil.delete_lambda_function(func_name)229    @pytest.mark.skipif(230        condition=not isinstance(231            lambda_api.LAMBDA_EXECUTOR, lambda_executors.LambdaExecutorReuseContainers232        ),233        reason="Test only applicable if docker-reuse executor is selected",234    )235    def test_destroy_idle_containers(self):236        executor = lambda_api.LAMBDA_EXECUTOR237        func_name = "test_destroy_idle_containers"238        func_arn = lambda_api.func_arn(func_name)239        # make sure existing containers are gone240        executor.destroy_existing_docker_containers()241        self.assertEqual(0, len(executor.get_all_container_names()))242        # deploy and invoke lambda without Docker243        testutil.create_lambda_function(244            func_name=func_name,245            handler_file=TEST_LAMBDA_ENV,246            libs=TEST_LAMBDA_LIBS,247            envvars={"Hello": "World"},248        )249        self.assertEqual(0, len(executor.get_all_container_names()))250        self.lambda_client.invoke(FunctionName=func_name, Payload=b"{}")251        self.assertEqual(1, len(executor.get_all_container_names()))252        # try to destroy idle containers.253        executor.idle_container_destroyer()254        self.assertEqual(1, len(executor.get_all_container_names()))255        # simulate an idle container256        executor.function_invoke_times[func_arn] = (257            int(time.time() * 1000) - lambda_executors.MAX_CONTAINER_IDLE_TIME_MS258        )259        executor.idle_container_destroyer()260        def assert_container_destroyed():261            self.assertEqual(0, len(executor.get_all_container_names()))262        retry(assert_container_destroyed, retries=3)263        # clean up264        testutil.delete_lambda_function(func_name)265class TestLocalExecutors(unittest.TestCase):266    def test_python3_runtime_multiple_create_with_conflicting_module(self):267        lambda_client = aws_stack.create_external_boto_client("lambda")268        original_do_use_docker = lambda_api.DO_USE_DOCKER269        try:270            # always use the local runner271            lambda_api.DO_USE_DOCKER = False272            python3_with_settings1 = load_file(TEST_LAMBDA_PYTHON3_MULTIPLE_CREATE1, mode="rb")273            python3_with_settings2 = load_file(TEST_LAMBDA_PYTHON3_MULTIPLE_CREATE2, mode="rb")274            lambda_name1 = "test1-%s" % short_uid()275            testutil.create_lambda_function(276                func_name=lambda_name1,277                zip_file=python3_with_settings1,278                runtime=LAMBDA_RUNTIME_PYTHON36,279                handler="handler1.handler",280            )281            lambda_name2 = "test2-%s" % short_uid()282            testutil.create_lambda_function(283                func_name=lambda_name2,284                zip_file=python3_with_settings2,285                runtime=LAMBDA_RUNTIME_PYTHON36,286                handler="handler2.handler",287            )288            result1 = lambda_client.invoke(FunctionName=lambda_name1, Payload=b"{}")289            result_data1 = result1["Payload"].read()290            result2 = lambda_client.invoke(FunctionName=lambda_name2, Payload=b"{}")291            result_data2 = result2["Payload"].read()292            self.assertEqual(200, result1["StatusCode"])293            self.assertIn("setting1", to_str(result_data1))294            self.assertEqual(200, result2["StatusCode"])295            self.assertIn("setting2", to_str(result_data2))296            # clean up...

notification.py

Source:notification.py

...7    def __init__(self, scope: core.Construct, component_id: str, **kwargs):8        super().__init__(scope, component_id, **kwargs)9        self.stack_name = self.node.try_get_context("prefix")10        self.component_id = component_id11        self._handler = self.create_lambda_function()12    def create_lambda_function(self) -> core.Resource:13        """14        Lambda Function to say hello.15        Ref: https://github.com/aws-samples/aws-cdk-examples/tree/master/python/lambda-cron16        """17        lambdaFn_id = f"{self.stack_name}-{self.component_id}-" + "lambda_handler"18        lambdaFn_path = str(pathlib.Path(__file__).resolve().parent) + "/lambdafn/sample_function/"19        lambdaFn = aws_lambda.Function(20            scope=self,21            id=lambdaFn_id,22            function_name=lambdaFn_id,23            code=aws_lambda.AssetCode(path=lambdaFn_path),24            handler="lambda_handler.lambda_handler",25            timeout=core.Duration.seconds(300),26            runtime=aws_lambda.Runtime.PYTHON_3_7,...

lambda_func.py

Source:lambda_func.py

1import boto32import json3#client = boto3.client('lambda')4def lambda_build(LambdaFunctionName, iamRole):5    client = boto3.client('lambda')6    create_lambda_function = client.create_function(7        FunctionName=LambdaFunctionName,8        Runtime='python3.7',9        Role=iamRole,10        Handler='ssm.lambda_handler',11        Description='Start a virtual machine',12	    Code = {'S3Bucket':'test-bucket-20201610', 'S3Key':'ssm.zip'},13        Timeout=1014        )15    return create_lambda_function...

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.