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

How to use requests_error_response_xml method in localstack

Best Python code snippet using localstack_python

aws_responses.py

Source:aws_responses.py

...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)...

route53_listener.py

Source:route53_listener.py

...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()...

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.