KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use delete_url_config method in localstack
Best Python code snippet using localstack_python
lambda_api.py
Source:lambda_api.py 
...1335 response = prev_url_config.copy()1336 response.pop("CustomId")1337 return response1338@app.route("%s/functions/<function>/url" % API_PATH_ROOT_2, methods=["DELETE"])1339def delete_url_config(function):1340 arn = func_arn(function)1341 qualifier = request.args.get("Qualifier")1342 q_arn = func_qualifier(function, qualifier)1343 arn = q_arn or arn1344 lambda_backend = LambdaRegion.get()1345 if arn not in lambda_backend.url_configs:1346 response = error_response("Function does not exist", 404, "ResourceNotFoundException")1347 return response1348 lambda_backend.url_configs.pop(arn)1349 return {}1350def add_permission_policy_statement(1351 resource_name, resource_arn, resource_arn_qualified, qualifier=None1352):1353 region = LambdaRegion.get()...
models.py
Source:models.py
...797 return alias798 def create_url_config(self, config):799 self.url_config = FunctionUrlConfig(function=self, config=config)800 return self.url_config801 def delete_url_config(self):802 self.url_config = None803 def get_url_config(self):804 if not self.url_config:805 raise FunctionUrlConfigNotFound()806 return self.url_config807 def update_url_config(self, config):808 self.url_config.update(config)809 return self.url_config810class FunctionUrlConfig:811 def __init__(self, function: LambdaFunction, config):812 self.function = function813 self.config = config814 self.url = f"https://{uuid.uuid4().hex}.lambda-url.{function.region}.on.aws"815 self.created = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")816 self.last_modified = self.created817 def to_dict(self):818 return {819 "FunctionUrl": self.url,820 "FunctionArn": self.function.function_arn,821 "AuthType": self.config.get("AuthType"),822 "Cors": self.config.get("Cors"),823 "CreationTime": self.created,824 "LastModifiedTime": self.last_modified,825 }826 def update(self, new_config):827 if new_config.get("Cors"):828 self.config["Cors"] = new_config["Cors"]829 if new_config.get("AuthType"):830 self.config["AuthType"] = new_config["AuthType"]831 self.last_modified = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")832class EventSourceMapping(CloudFormationModel):833 def __init__(self, spec):834 # required835 self.function_name = spec["FunctionName"]836 self.event_source_arn = spec["EventSourceArn"]837 # optional838 self.batch_size = spec.get("BatchSize")839 self.starting_position = spec.get("StartingPosition", "TRIM_HORIZON")840 self.enabled = spec.get("Enabled", True)841 self.starting_position_timestamp = spec.get("StartingPositionTimestamp", None)842 self.function_arn = spec["FunctionArn"]843 self.uuid = str(uuid.uuid4())844 self.last_modified = time.mktime(datetime.datetime.utcnow().timetuple())845 def _get_service_source_from_arn(self, event_source_arn):846 return event_source_arn.split(":")[2].lower()847 def _validate_event_source(self, event_source_arn):848 valid_services = ("dynamodb", "kinesis", "sqs")849 service = self._get_service_source_from_arn(event_source_arn)850 return True if service in valid_services else False851 @property852 def event_source_arn(self):853 return self._event_source_arn854 @event_source_arn.setter855 def event_source_arn(self, event_source_arn):856 if not self._validate_event_source(event_source_arn):857 raise ValueError(858 "InvalidParameterValueException", "Unsupported event source type"859 )860 self._event_source_arn = event_source_arn861 @property862 def batch_size(self):863 return self._batch_size864 @batch_size.setter865 def batch_size(self, batch_size):866 batch_size_service_map = {867 "kinesis": (100, 10000),868 "dynamodb": (100, 1000),869 "sqs": (10, 10),870 }871 source_type = self._get_service_source_from_arn(self.event_source_arn)872 batch_size_for_source = batch_size_service_map[source_type]873 if batch_size is None:874 self._batch_size = batch_size_for_source[0]875 elif batch_size > batch_size_for_source[1]:876 error_message = "BatchSize {} exceeds the max of {}".format(877 batch_size, batch_size_for_source[1]878 )879 raise ValueError("InvalidParameterValueException", error_message)880 else:881 self._batch_size = int(batch_size)882 def get_configuration(self):883 return {884 "UUID": self.uuid,885 "BatchSize": self.batch_size,886 "EventSourceArn": self.event_source_arn,887 "FunctionArn": self.function_arn,888 "LastModified": self.last_modified,889 "LastProcessingResult": "",890 "State": "Enabled" if self.enabled else "Disabled",891 "StateTransitionReason": "User initiated",892 }893 def delete(self, account_id, region_name):894 lambda_backend = lambda_backends[account_id][region_name]895 lambda_backend.delete_event_source_mapping(self.uuid)896 @staticmethod897 def cloudformation_name_type():898 return None899 @staticmethod900 def cloudformation_type():901 # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html902 return "AWS::Lambda::EventSourceMapping"903 @classmethod904 def create_from_cloudformation_json(905 cls, resource_name, cloudformation_json, account_id, region_name, **kwargs906 ):907 properties = cloudformation_json["Properties"]908 lambda_backend = lambda_backends[account_id][region_name]909 return lambda_backend.create_event_source_mapping(properties)910 @classmethod911 def update_from_cloudformation_json(912 cls,913 original_resource,914 new_resource_name,915 cloudformation_json,916 account_id,917 region_name,918 ):919 properties = cloudformation_json["Properties"]920 event_source_uuid = original_resource.uuid921 lambda_backend = lambda_backends[account_id][region_name]922 return lambda_backend.update_event_source_mapping(event_source_uuid, properties)923 @classmethod924 def delete_from_cloudformation_json(925 cls, resource_name, cloudformation_json, account_id, region_name926 ):927 properties = cloudformation_json["Properties"]928 lambda_backend = lambda_backends[account_id][region_name]929 esms = lambda_backend.list_event_source_mappings(930 event_source_arn=properties["EventSourceArn"],931 function_name=properties["FunctionName"],932 )933 for esm in esms:934 if esm.uuid == resource_name:935 esm.delete(account_id, region_name)936 @property937 def physical_resource_id(self):938 return self.uuid939class LambdaVersion(CloudFormationModel):940 def __init__(self, spec):941 self.version = spec["Version"]942 def __repr__(self):943 return str(self.logical_resource_id)944 @staticmethod945 def cloudformation_name_type():946 return None947 @staticmethod948 def cloudformation_type():949 # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-version.html950 return "AWS::Lambda::Version"951 @classmethod952 def create_from_cloudformation_json(953 cls, resource_name, cloudformation_json, account_id, region_name, **kwargs954 ):955 properties = cloudformation_json["Properties"]956 function_name = properties["FunctionName"]957 func = lambda_backends[account_id][region_name].publish_function(function_name)958 spec = {"Version": func.version}959 return LambdaVersion(spec)960class LambdaStorage(object):961 def __init__(self, region_name, account_id):962 # Format 'func_name' {'versions': []}963 self._functions = {}964 self._aliases = dict()965 self._arns = weakref.WeakValueDictionary()966 self.region_name = region_name967 self.account_id = account_id968 def _get_latest(self, name):969 return self._functions[name]["latest"]970 def _get_version(self, name, version):971 index = version - 1972 try:973 return self._functions[name]["versions"][index]974 except IndexError:975 return None976 def delete_alias(self, name, function_name):977 fn = self.get_function_by_name_or_arn(function_name)978 return fn.delete_alias(name)979 def get_alias(self, name, function_name):980 fn = self.get_function_by_name_or_arn(function_name)981 return fn.get_alias(name)982 def put_alias(983 self, name, function_name, function_version, description, routing_config984 ):985 fn = self.get_function_by_name_or_arn(function_name)986 return fn.put_alias(name, description, function_version, routing_config)987 def update_alias(988 self, name, function_name, function_version, description, routing_config989 ):990 fn = self.get_function_by_name_or_arn(function_name)991 return fn.update_alias(name, description, function_version, routing_config)992 def get_function_by_name(self, name, qualifier=None):993 if name not in self._functions:994 return None995 if qualifier is None:996 return self._get_latest(name)997 try:998 return self._get_version(name, int(qualifier))999 except ValueError:1000 return self._functions[name]["latest"]1001 def list_versions_by_function(self, name):1002 if name not in self._functions:1003 return None1004 latest = copy.copy(self._functions[name]["latest"])1005 latest.function_arn += ":$LATEST"1006 return [latest] + self._functions[name]["versions"]1007 def get_arn(self, arn):1008 # Function ARN may contain an alias1009 # arn:aws:lambda:region:account_id:function:<fn_name>:<alias_name>1010 if ":" in arn.split(":function:")[-1]:1011 # arn = arn:aws:lambda:region:account_id:function:<fn_name>1012 arn = ":".join(arn.split(":")[0:-1])1013 return self._arns.get(arn, None)1014 def get_function_by_name_or_arn(1015 self, name_or_arn, qualifier=None1016 ) -> LambdaFunction:1017 fn = self.get_function_by_name(name_or_arn, qualifier) or self.get_arn(1018 name_or_arn1019 )1020 if fn is None:1021 if name_or_arn.startswith("arn:aws"):1022 arn = name_or_arn1023 else:1024 arn = make_function_arn(self.region_name, self.account_id, name_or_arn)1025 if qualifier:1026 arn = f"{arn}:{qualifier}"1027 raise UnknownFunctionException(arn)1028 return fn1029 def put_function(self, fn):1030 """1031 :param fn: Function1032 :type fn: LambdaFunction1033 """1034 valid_role = re.match(InvalidRoleFormat.pattern, fn.role)1035 if valid_role:1036 account = valid_role.group(2)1037 if account != self.account_id:1038 raise CrossAccountNotAllowed()1039 try:1040 iam_backend = iam_backends[self.account_id]["global"]1041 iam_backend.get_role_by_arn(fn.role)1042 except IAMNotFoundException:1043 raise InvalidParameterValueException(1044 "The role defined for the function cannot be assumed by Lambda."1045 )1046 else:1047 raise InvalidRoleFormat(fn.role)1048 if fn.function_name in self._functions:1049 self._functions[fn.function_name]["latest"] = fn1050 else:1051 self._functions[fn.function_name] = {"latest": fn, "versions": []}1052 # instantiate a new policy for this version of the lambda1053 fn.policy = Policy(fn)1054 self._arns[fn.function_arn] = fn1055 def publish_function(self, name_or_arn, description=""):1056 function = self.get_function_by_name_or_arn(name_or_arn)1057 name = function.function_name1058 if name not in self._functions:1059 return None1060 if not self._functions[name]["latest"]:1061 return None1062 new_version = len(self._functions[name]["versions"]) + 11063 fn = copy.copy(self._functions[name]["latest"])1064 fn.set_version(new_version)1065 if description:1066 fn.description = description1067 self._functions[name]["versions"].append(fn)1068 self._arns[fn.function_arn] = fn1069 return fn1070 def del_function(self, name_or_arn, qualifier=None):1071 function = self.get_function_by_name_or_arn(name_or_arn, qualifier)1072 name = function.function_name1073 if not qualifier:1074 # Something is still reffing this so delete all arns1075 latest = self._functions[name]["latest"].function_arn1076 del self._arns[latest]1077 for fn in self._functions[name]["versions"]:1078 del self._arns[fn.function_arn]1079 del self._functions[name]1080 elif qualifier == "$LATEST":1081 self._functions[name]["latest"] = None1082 # If theres no functions left1083 if (1084 not self._functions[name]["versions"]1085 and not self._functions[name]["latest"]1086 ):1087 del self._functions[name]1088 else:1089 fn = self.get_function_by_name(name, qualifier)1090 if fn:1091 self._functions[name]["versions"].remove(fn)1092 # If theres no functions left1093 if (1094 not self._functions[name]["versions"]1095 and not self._functions[name]["latest"]1096 ):1097 del self._functions[name]1098 def all(self):1099 result = []1100 for function_group in self._functions.values():1101 latest = copy.deepcopy(function_group["latest"])1102 latest.function_arn = "{}:$LATEST".format(latest.function_arn)1103 result.append(latest)1104 result.extend(function_group["versions"])1105 return result1106 def latest(self):1107 """1108 Return the list of functions with version @LATEST1109 :return:1110 """1111 result = []1112 for function_group in self._functions.values():1113 if function_group["latest"] is not None:1114 result.append(function_group["latest"])1115 return result1116class LayerStorage(object):1117 def __init__(self):1118 self._layers = {}1119 self._arns = weakref.WeakValueDictionary()1120 def put_layer_version(self, layer_version):1121 """1122 :param layer_version: LayerVersion1123 """1124 if layer_version.name not in self._layers:1125 self._layers[layer_version.name] = Layer(layer_version)1126 self._layers[layer_version.name].attach_version(layer_version)1127 def list_layers(self):1128 return [1129 layer.to_dict() for layer in self._layers.values() if layer.layer_versions1130 ]1131 def delete_layer_version(self, layer_name, layer_version):1132 self._layers[layer_name].delete_version(layer_version)1133 def get_layer_version(self, layer_name, layer_version):1134 if layer_name not in self._layers:1135 raise UnknownLayerException()1136 for lv in self._layers[layer_name].layer_versions.values():1137 if lv.version == int(layer_version):1138 return lv1139 raise UnknownLayerException()1140 def get_layer_versions(self, layer_name):1141 if layer_name in self._layers:1142 return list(iter(self._layers[layer_name].layer_versions.values()))1143 return []1144 def get_layer_version_by_arn(self, layer_version_arn):1145 split_arn = split_layer_arn(layer_version_arn)1146 if split_arn.layer_name in self._layers:1147 return self._layers[split_arn.layer_name].layer_versions.get(1148 split_arn.version, None1149 )1150 return None1151class LambdaBackend(BaseBackend):1152 """1153 Implementation of the AWS Lambda endpoint.1154 Invoking functions is supported - they will run inside a Docker container, emulating the real AWS behaviour as closely as possible.1155 It is possible to connect from AWS Lambdas to other services, as long as you are running Moto in ServerMode.1156 The Lambda has access to environment variables `MOTO_HOST` and `MOTO_PORT`, which can be used to build the url that MotoServer runs on:1157 .. sourcecode:: python1158 def lambda_handler(event, context):1159 host = os.environ.get("MOTO_HOST")1160 port = os.environ.get("MOTO_PORT")1161 url = host + ":" + port1162 ec2 = boto3.client('ec2', region_name='us-west-2', endpoint_url=url)1163 # Or even simpler:1164 full_url = os.environ.get("MOTO_HTTP_ENDPOINT")1165 ec2 = boto3.client("ec2", region_name="eu-west-1", endpoint_url=full_url)1166 ec2.do_whatever_inside_the_existing_moto_server()1167 Moto will run on port 5000 by default. This can be overwritten by setting an environment variable when starting Moto:1168 .. sourcecode:: bash1169 # This env var will be propagated to the Docker container running the Lambda functions1170 MOTO_PORT=5000 moto_server1171 The Docker container uses the default network mode, `bridge`.1172 The following environment variables are available for fine-grained control over the Docker connection options:1173 .. sourcecode:: bash1174 # Provide the name of a custom network to connect to1175 MOTO_DOCKER_NETWORK_NAME=mycustomnetwork moto_server1176 # Override the network mode1177 # For example, network_mode=host would use the network of the host machine1178 # Note that this option will be ignored if MOTO_DOCKER_NETWORK_NAME is also set1179 MOTO_DOCKER_NETWORK_MODE=host moto_server1180 The Docker images used by Moto are taken from the `lambci/lambda`-repo by default. Use the following environment variable to configure a different repo:1181 .. sourcecode:: bash1182 MOTO_DOCKER_LAMBDA_IMAGE=mLupin/docker-lambda1183 .. note:: When using the decorators, a Docker container cannot reach Moto, as it does not run as a server. Any boto3-invocations used within your Lambda will try to connect to AWS.1184 """1185 def __init__(self, region_name, account_id):1186 super().__init__(region_name, account_id)1187 self._lambdas = LambdaStorage(region_name=region_name, account_id=account_id)1188 self._event_source_mappings = {}1189 self._layers = LayerStorage()1190 @staticmethod1191 def default_vpc_endpoint_service(service_region, zones):1192 """Default VPC endpoint service."""1193 return BaseBackend.default_vpc_endpoint_service_factory(1194 service_region, zones, "lambda"1195 )1196 def create_alias(1197 self, name, function_name, function_version, description, routing_config1198 ):1199 return self._lambdas.put_alias(1200 name, function_name, function_version, description, routing_config1201 )1202 def delete_alias(self, name, function_name):1203 return self._lambdas.delete_alias(name, function_name)1204 def get_alias(self, name, function_name):1205 return self._lambdas.get_alias(name, function_name)1206 def update_alias(1207 self, name, function_name, function_version, description, routing_config1208 ):1209 """1210 The RevisionId parameter is not yet implemented1211 """1212 return self._lambdas.update_alias(1213 name, function_name, function_version, description, routing_config1214 )1215 def create_function(self, spec):1216 function_name = spec.get("FunctionName", None)1217 if function_name is None:1218 raise RESTError("InvalidParameterValueException", "Missing FunctionName")1219 fn = LambdaFunction(1220 account_id=self.account_id,1221 spec=spec,1222 region=self.region_name,1223 version="$LATEST",1224 )1225 self._lambdas.put_function(fn)1226 if spec.get("Publish"):1227 ver = self.publish_function(function_name)1228 fn = copy.deepcopy(1229 fn1230 ) # We don't want to change the actual version - just the return value1231 fn.version = ver.version1232 return fn1233 def create_function_url_config(self, name_or_arn, config):1234 """1235 The Qualifier-parameter is not yet implemented.1236 Function URLs are not yet mocked, so invoking them will fail1237 """1238 function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1239 return function.create_url_config(config)1240 def delete_function_url_config(self, name_or_arn):1241 """1242 The Qualifier-parameter is not yet implemented1243 """1244 function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1245 function.delete_url_config()1246 def get_function_url_config(self, name_or_arn):1247 """1248 The Qualifier-parameter is not yet implemented1249 """1250 function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1251 if not function:1252 raise UnknownFunctionException(arn=name_or_arn)1253 return function.get_url_config()1254 def update_function_url_config(self, name_or_arn, config):1255 """1256 The Qualifier-parameter is not yet implemented1257 """1258 function = self._lambdas.get_function_by_name_or_arn(name_or_arn)1259 return function.update_url_config(config)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
