How to use requests_error_response_xml method in localstack

Best Python code snippet using localstack_python

aws_responses.py

Source:aws_responses.py Github

copy

Full Screen

...26 return Response(json.dumps(result), status=code, headers=headers)27def requests_error_response_json(message, code=500, error_type='InternalFailure'):28 response = flask_error_response_json(message, code=code, error_type=error_type)29 return flask_to_requests_response(response)30def requests_error_response_xml(message, code=400, code_string='InvalidParameter', service=None, xmlns=None):31 response = RequestsResponse()32 xmlns = xmlns or 'http://%s.amazonaws.com/doc/2010-03-31/' % service33 response._content = """<ErrorResponse xmlns="{xmlns}"><Error>34 <Type>Sender</Type>35 <Code>{code_string}</Code>36 <Message>{message}</Message>37 </Error><RequestId>{req_id}</RequestId>38 </ErrorResponse>""".format(xmlns=xmlns, message=message, code_string=code_string, req_id=short_uid())39 response.status_code = code40 return response41def requests_response_xml(action, response, xmlns=None, service=None):42 xmlns = xmlns or 'http://%s.amazonaws.com/doc/2010-03-31/' % service43 response = json_safe(response)44 response = {'{action}Result'.format(action=action): response}45 response = xmltodict.unparse(response)46 if response.startswith('<?xml'):47 response = re.sub(r'<\?xml [^\?]+\?>', '', response)48 result = ("""49 <{action}Response xmlns="{xmlns}">50 {response}51 </{action}Response>52 """).strip()53 result = result.format(action=action, xmlns=xmlns, response=response)54 result = requests_response(result)55 return result56def requests_error_response_xml_signature_calculation(message, string_to_sign=None, signature=None, expires=None,57 code=400, code_string='AccessDenied', aws_access_token='temp'):58 response = RequestsResponse()59 response_template = """<?xml version="1.0" encoding="UTF-8"?>60 <Error>61 <Code>{code_string}</Code>62 <Message>{message}</Message>63 <RequestId>{req_id}</RequestId>64 <HostId>{host_id}</HostId>65 </Error>""".format(message=message, code_string=code_string, req_id=short_uid(), host_id=short_uid())66 parsed_response = xmltodict.parse(response_template)67 response.status_code = code68 if signature and string_to_sign or code_string == 'SignatureDoesNotMatch':69 bytes_signature = binascii.hexlify(bytes(signature, encoding='utf-8'))70 parsed_response['Error']['Code'] = code_string71 parsed_response['Error']['AWSAccessKeyId'] = aws_access_token72 parsed_response['Error']['StringToSign'] = string_to_sign73 parsed_response['Error']['SignatureProvided'] = signature74 parsed_response['Error']['StringToSignBytes'] = '{}'.format(bytes_signature.decode('utf-8'))75 response._content = xmltodict.unparse(parsed_response)76 response.headers['Content-Length'] = str(len(response._content))77 if expires and code_string == 'AccessDenied':78 server_time = datetime.datetime.utcnow().isoformat()[:-4]79 expires_isoformat = datetime.datetime.fromtimestamp(int(expires)).isoformat()[:-4]80 parsed_response['Error']['Code'] = code_string81 parsed_response['Error']['Expires'] = '{}Z'.format(expires_isoformat)82 parsed_response['Error']['ServerTime'] = '{}Z'.format(server_time)83 response._content = xmltodict.unparse(parsed_response)84 response.headers['Content-Length'] = str(len(response._content))85 if not signature and not expires and code_string == 'AccessDenied':86 response._content = xmltodict.unparse(parsed_response)87 response.headers['Content-Length'] = str(len(response._content))88 if response._content:89 return response90def flask_error_response_xml(message, code=500, code_string='InternalFailure', service=None, xmlns=None):91 response = requests_error_response_xml(message, code=code, code_string=code_string, service=service, xmlns=xmlns)92 return requests_to_flask_response(response)93def requests_error_response(req_headers, message, code=500, error_type='InternalFailure', service=None, xmlns=None):94 ctype = req_headers.get('Content-Type', '')95 accept = req_headers.get('Accept', '')96 is_json = 'json' in ctype or 'json' in accept97 if is_json:98 return requests_error_response_json(message=message, code=code, error_type=error_type)99 return requests_error_response_xml(message, code=code, code_string=error_type, service=service, xmlns=xmlns)100def requests_response(content, status_code=200, headers={}):101 resp = RequestsResponse()102 content = json.dumps(content) if isinstance(content, dict) else content103 resp._content = content104 resp.status_code = status_code105 resp.headers = headers106 return resp107def flask_to_requests_response(r):108 return requests_response(r.data, status_code=r.status_code, headers=r.headers)109def requests_to_flask_response(r):110 return Response(r.content, status=r.status_code, headers=dict(r.headers))111def response_regex_replace(response, search, replace):112 content = re.sub(search, replace, to_str(response._content), flags=re.DOTALL | re.MULTILINE)113 set_response_content(response, content)...

Full Screen

Full Screen

route53_listener.py

Source:route53_listener.py Github

copy

Full Screen

...78 if method == "GET":79 if set_id:80 result = region_details.reusable_delegation_sets.get(set_id)81 if not result:82 return requests_error_response_xml(83 "Delegation set not found", code_string="NoSuchDelegationSet"84 )85 return {"GetReusableDelegationSetResponse": {"DelegationSet": result}}86 reusable_sets_list = list(region_details.reusable_delegation_sets.values())87 result = {88 "ListReusableDelegationSetsResponse": {89 "IsTruncated": False,90 }91 }92 if len(reusable_sets_list) > 0:93 result["ListReusableDelegationSetsResponse"]["DelegationSets"] = {94 "DelegationSet": reusable_sets_list95 }96 return result97 if method == "POST":98 req_data = xmltodict.parse(to_str(data))99 req_data = req_data.get("CreateReusableDelegationSetRequest")100 req_data.pop("@xmlns", None)101 req_data["Id"] = short_uid()102 region_details.reusable_delegation_sets[req_data["Id"]] = req_data103 return {"CreateReusableDelegationSetResponse": {"DelegationSet": req_data}}104 if method == "DELETE":105 existing = region_details.reusable_delegation_sets.pop(set_id, None)106 if not existing:107 return requests_error_response_xml(108 "Delegation set not found", code_string="NoSuchDelegationSet"109 )110 return {"DeleteReusableDelegationSetResponse": {}}111def handle_hosted_zones_by_vpc_request(method: str, path: str, data: Dict):112 def _zone(z):113 zone_id = z["HostedZoneId"]114 hosted_zone = client.get_hosted_zone(Id=zone_id).get("HostedZone", {})115 result = {116 "HostedZoneId": zone_id,117 "Name": hosted_zone.get("Name"),118 "Owner": {"OwningAccount": constants.TEST_AWS_ACCOUNT_ID},119 }120 return result121 region_details = Route53Backend.get()...

Full Screen

Full Screen

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:

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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful