KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use delete_function_url_config method in localstack

Best Python code snippet using localstack_python

awslambda.py

Source:awslambda.py

1import os2from localstack.services.awslambda.lambda_api import get_lambda_policy_name3from localstack.services.awslambda.lambda_utils import get_handler_file_from_name4from localstack.services.cloudformation.deployment_utils import (5    generate_default_name,6    get_cfn_response_mod_file,7    select_parameters,8)9from localstack.services.cloudformation.service_models import LOG, REF_ID_ATTRS, GenericBaseModel10from localstack.utils.aws import aws_stack11from localstack.utils.common import (12    cp_r,13    is_base64,14    is_zip_file,15    mkdir,16    new_tmp_dir,17    rm_rf,18    save_file,19    select_attributes,20    short_uid,21    to_bytes,22)23from localstack.utils.testutil import create_zip_file24class LambdaFunction(GenericBaseModel):25    @staticmethod26    def cloudformation_type():27        return "AWS::Lambda::Function"28    def fetch_state(self, stack_name, resources):29        func_name = self.resolve_refs_recursively(stack_name, self.props["FunctionName"], resources)30        return aws_stack.connect_to_service("lambda").get_function(FunctionName=func_name)31    def get_physical_resource_id(self, attribute=None, **kwargs):32        func_name = self.props.get("FunctionName")33        if attribute == "Arn":34            return aws_stack.lambda_function_arn(func_name)35        return func_name36    def update_resource(self, new_resource, stack_name, resources):37        props = new_resource["Properties"]38        client = aws_stack.connect_to_service("lambda")39        config_keys = [40            "Description",41            "Environment",42            "FunctionName",43            "Handler",44            "ImageConfig",45            "Layers",46            "MemorySize",47            "Role",48            "Runtime",49            "Timeout",50            "TracingConfig",51            "VpcConfig",52        ]53        update_config_props = select_attributes(props, config_keys)54        update_config_props = self.resolve_refs_recursively(55            stack_name, update_config_props, resources56        )57        if "Timeout" in update_config_props:58            update_config_props["Timeout"] = int(update_config_props["Timeout"])59        if "Code" in props:60            code = props["Code"] or {}61            if not code.get("ZipFile"):62                LOG.debug(63                    'Updating code for Lambda "%s" from location: %s', props["FunctionName"], code64                )65            code = LambdaFunction.get_lambda_code_param(props, _include_arch=True)66            client.update_function_code(FunctionName=props["FunctionName"], **code)67        if "Environment" in update_config_props:68            environment_variables = update_config_props["Environment"].get("Variables", {})69            update_config_props["Environment"]["Variables"] = {70                k: str(v) for k, v in environment_variables.items()71            }72        return client.update_function_configuration(**update_config_props)73    @staticmethod74    def add_defaults(resource, stack_name: str):75        func_name = resource.get("Properties", {}).get("FunctionName")76        if not func_name:77            resource["Properties"]["FunctionName"] = generate_default_name(78                stack_name, resource["LogicalResourceId"]79            )80    @staticmethod81    def get_lambda_code_param(params, _include_arch=False, **kwargs):82        code = params.get("Code", {})83        zip_file = code.get("ZipFile")84        if zip_file and not is_base64(zip_file) and not is_zip_file(to_bytes(zip_file)):85            tmp_dir = new_tmp_dir()86            handler_file = get_handler_file_from_name(params["Handler"], runtime=params["Runtime"])87            tmp_file = os.path.join(tmp_dir, handler_file)88            save_file(tmp_file, zip_file)89            # add 'cfn-response' module to archive - see:90            # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/cfn-lambda-function-code-cfnresponsemodule.html91            cfn_response_tmp_file = get_cfn_response_mod_file()92            cfn_response_mod_dir = os.path.join(tmp_dir, "node_modules", "cfn-response")93            mkdir(cfn_response_mod_dir)94            cp_r(95                cfn_response_tmp_file,96                os.path.join(cfn_response_mod_dir, "index.js"),97            )98            # create zip file99            zip_file = create_zip_file(tmp_dir, get_content=True)100            code["ZipFile"] = zip_file101            rm_rf(tmp_dir)102        if _include_arch and "Architectures" in params:103            code["Architectures"] = params.get("Architectures")104        return code105    @staticmethod106    def get_deploy_templates():107        def get_delete_params(params, **kwargs):108            return {"FunctionName": params.get("FunctionName")}109        def get_environment_params(params, **kwargs):110            # botocore/data/lambda/2015-03-31/service-2.json:1161 (EnvironmentVariableValue)111            # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-lambda-function-environment.html112            if "Environment" in params:113                environment_variables = params["Environment"].get("Variables", {})114                return {"Variables": {k: str(v) for k, v in environment_variables.items()}}115        return {116            "create": {117                "function": "create_function",118                "parameters": {119                    "Architectures": "Architectures",120                    "Code": LambdaFunction.get_lambda_code_param,121                    "Description": "Description",122                    "Environment": get_environment_params,123                    "FunctionName": "FunctionName",124                    "Handler": "Handler",125                    "ImageConfig": "ImageConfig",126                    "PackageType": "PackageType",127                    "Layers": "Layers",128                    "MemorySize": "MemorySize",129                    "Runtime": "Runtime",130                    "Role": "Role",131                    "Timeout": "Timeout",132                    "TracingConfig": "TracingConfig",133                    "VpcConfig": "VpcConfig"134                    # TODO add missing fields135                },136                "defaults": {"Role": "test_role"},137                "types": {"Timeout": int, "MemorySize": int},138            },139            "delete": {"function": "delete_function", "parameters": get_delete_params},140        }141class LambdaFunctionVersion(GenericBaseModel):142    @staticmethod143    def cloudformation_type():144        return "AWS::Lambda::Version"145    def fetch_state(self, stack_name, resources):146        name = self.resolve_refs_recursively(stack_name, self.props.get("FunctionName"), resources)147        if not name:148            return None149        func_name = aws_stack.lambda_function_name(name)150        func_version = name.split(":")[7] if len(name.split(":")) > 7 else "$LATEST"151        versions = aws_stack.connect_to_service("lambda").list_versions_by_function(152            FunctionName=func_name153        )154        return ([v for v in versions["Versions"] if v["Version"] == func_version] or [None])[0]155    @staticmethod156    def get_deploy_templates():157        return {158            "create": {159                "function": "publish_version",160                "parameters": select_parameters("FunctionName", "CodeSha256", "Description"),161            }162        }163class LambdaEventSourceMapping(GenericBaseModel):164    @staticmethod165    def cloudformation_type():166        return "AWS::Lambda::EventSourceMapping"167    def fetch_state(self, stack_name, resources):168        props = self.props169        source_arn = props.get("EventSourceArn")170        self_managed_src = props.get("SelfManagedEventSource")171        function_name = self.resolve_refs_recursively(stack_name, props["FunctionName"], resources)172        source_arn = self.resolve_refs_recursively(stack_name, source_arn, resources)173        if not function_name or (not source_arn and not self_managed_src):174            raise Exception("ResourceNotFound")175        def _matches(m):176            return m["FunctionArn"] == lambda_arn and (177                m.get("EventSourceArn") == source_arn178                or m.get("SelfManagedEventSource") == self_managed_src179            )180        client = aws_stack.connect_to_service("lambda")181        lambda_arn = client.get_function(FunctionName=function_name)["Configuration"]["FunctionArn"]182        kwargs = {"EventSourceArn": source_arn} if source_arn else {}183        mappings = client.list_event_source_mappings(FunctionName=function_name, **kwargs)184        mapping = list(filter(lambda m: _matches(m), mappings["EventSourceMappings"]))185        if not mapping:186            raise Exception("ResourceNotFound")187        return mapping[0]188    def get_cfn_attribute(self, attribute_name):189        if attribute_name in REF_ID_ATTRS:190            return self.props.get("UUID")191        return super(LambdaEventSourceMapping, self).get_cfn_attribute(attribute_name)192    def get_physical_resource_id(self, attribute=None, **kwargs):193        return self.props.get("UUID")194    @staticmethod195    def get_deploy_templates():196        return {"create": {"function": "create_event_source_mapping"}}197class LambdaPermission(GenericBaseModel):198    @staticmethod199    def cloudformation_type():200        return "AWS::Lambda::Permission"201    def fetch_state(self, stack_name, resources):202        props = self.props203        func_name = self.resolve_refs_recursively(stack_name, props.get("FunctionName"), resources)204        func_arn = aws_stack.lambda_function_arn(func_name)205        return self.do_fetch_state(func_name, func_arn)206    def do_fetch_state(self, resource_name, resource_arn):207        iam = aws_stack.connect_to_service("iam")208        props = self.props209        policy_name = get_lambda_policy_name(resource_name)210        policy_arn = aws_stack.policy_arn(policy_name)211        policy = iam.get_policy(PolicyArn=policy_arn)["Policy"]212        version = policy.get("DefaultVersionId")213        policy = iam.get_policy_version(PolicyArn=policy_arn, VersionId=version)["PolicyVersion"]214        statements = policy["Document"]["Statement"]215        statements = statements if isinstance(statements, list) else [statements]216        principal = props.get("Principal")217        existing = [218            s219            for s in statements220            if s["Action"] == props["Action"]221            and s["Resource"] == resource_arn222            and (223                not principal224                or s["Principal"] in [principal, {"Service": principal}, {"Service": [principal]}]225            )226        ]227        return existing[0] if existing else None228    def get_physical_resource_id(self, attribute=None, **kwargs):229        # return statement ID here to indicate that the resource has been deployed230        return self.props.get("Sid")231    @staticmethod232    def get_deploy_templates():233        def lambda_permission_params(params, **kwargs):234            result = select_parameters("FunctionName", "Action", "Principal")(params, **kwargs)235            result["StatementId"] = short_uid()236            return result237        return {238            "create": {239                "function": "add_permission",240                "parameters": lambda_permission_params,241            }242        }243class LambdaEventInvokeConfig(GenericBaseModel):244    @staticmethod245    def cloudformation_type():246        return "AWS::Lambda::EventInvokeConfig"247    def fetch_state(self, stack_name, resources):248        client = aws_stack.connect_to_service("lambda")249        props = self.props250        result = client.get_function_event_invoke_config(251            FunctionName=props.get("FunctionName"),252            Qualifier=props.get("FunctionName", "$LATEST"),253        )254        return result255    def get_physical_resource_id(self, attribute=None, **kwargs):256        props = self.props257        return "lambdaconfig-%s-%s" % (258            props.get("FunctionName"),259            props.get("Qualifier"),260        )261    @staticmethod262    def get_deploy_templates():263        return {264            "create": {"function": "put_function_event_invoke_config"},265            "delete": {266                "function": "delete_function_event_invoke_config",267                "parameters": {268                    "FunctionName": "FunctionName",269                    "Qualifier": "Qualifier",270                },271            },272        }273class LambdaUrl(GenericBaseModel):274    @classmethod275    def cloudformation_type(cls):276        return "AWS::Lambda::Url"277    def get_physical_resource_id(self, attribute=None, **kwargs):278        return self.props.get(279            "TargetFunctionArn"280        )  # TODO: if this isn't an ARN we need to resolve the full ARN here281    def fetch_state(self, stack_name, resources):282        client = aws_stack.connect_to_service("lambda")283        kwargs = {"FunctionName": self.props.get("TargetFunctionArn")}284        qualifier = self.props.get("Qualifier")285        if qualifier:286            kwargs["Qualifier"] = qualifier287        return client.get_function_url_config(**kwargs)288    def get_cfn_attribute(self, attribute_name):289        if attribute_name == "FunctionArn":290            return self.props.get("TargetFunctionArn")291        if attribute_name == "FunctionUrl":292            client = aws_stack.connect_to_service("lambda")293            url_config = client.get_function_url_config(294                FunctionName=self.props.get("TargetFunctionArn"),295                Qualifier=self.props.get("Qualifier", "$LATEST"),296            )297            return url_config["FunctionUrl"]298        return super(LambdaUrl, self).get_cfn_attribute(attribute_name)299    @staticmethod300    def get_deploy_templates():301        return {302            "create": {303                "function": "create_function_url_config",304                "parameters": {305                    "Qualifier": "Qualifier",306                    "Cors": "Cors",307                    "FunctionName": "TargetFunctionArn",308                    "AuthType": "AuthType",309                },310            },311            "delete": {312                "function": "delete_function_url_config",313                "parameters": {"FunctionName": "TargetFunctionArn", "Qualifier": "Qualifier"},314            },...

test_lambda_function_urls.py

Source:test_lambda_function_urls.py

...90    )91    resp.should.have.key("Cors").equals({"AllowCredentials": False})92@mock_lambda93@pytest.mark.parametrize("key", ["FunctionName", "FunctionArn"])94def test_delete_function_url_config(key):95    client = boto3.client("lambda", "us-east-2")96    function_name = str(uuid4())[0:6]97    fxn = client.create_function(98        FunctionName=function_name,99        Runtime="python3.7",100        Role=get_role_name(),101        Handler="lambda_function.lambda_handler",102        Code={"ZipFile": get_test_zip_file1()},103    )104    name_or_arn = fxn[key]105    client.create_function_url_config(AuthType="AWS_IAM", FunctionName=name_or_arn)106    client.delete_function_url_config(FunctionName=name_or_arn)107    # It no longer exists108    with pytest.raises(ClientError):...

_manager.py

Source:_manager.py

...16        self._create_lambda_function(name, role, handler_name, zip_file)17        function_url = self._create_lambda_function_public_url(name)18        return LambdaFunctionMetadata(name, function_url, role)19    def delete(self, name: str):20        self._client.delete_function_url_config(21            FunctionName=name,22        )23        self._client.delete_function(24            FunctionName=name,25        )26    def get_lambda_function(self, name: str) -> Optional[LambdaFunctionMetadata]:27        try:28            response = self._client.get_function(29                FunctionName=name,30            )31        except self._client.exceptions.ResourceNotFoundException:32            return None33        role = get_role_from_arn(arn=response["Configuration"]["Role"])34        response = self._client.get_function_url_config(FunctionName=name)...

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.