KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use make_accepted_response method in localstack

Best Python code snippet using localstack_python
helpers.py
Source:helpers.py
...75    if code == 404 and not error_type:76        error_type = "NotFoundException"77    error_type = error_type or "InvalidRequest"78    return requests_error_response_json(message, code=code, error_type=error_type)79def make_accepted_response():80    response = Response()81    response.status_code = 20282    return response83def get_api_id_from_path(path):84    match = re.match(PATH_REGEX_SUB, path)85    if match:86        return match.group(1)87    return re.match(PATH_REGEX_MAIN, path).group(1)88# -------------89# ACCOUNT APIs90# -------------91def get_account():92    region_details = APIGatewayRegion.get()93    return to_account_response_json(region_details.account)94def update_account(data):95    region_details = APIGatewayRegion.get()96    apply_json_patch_safe(region_details.account, data["patchOperations"], in_place=True)97    return to_account_response_json(region_details.account)98def handle_accounts(method, path, data, headers):99    if method == "GET":100        return get_account()101    if method == "PATCH":102        return update_account(data)103    return make_error_response("Not implemented for API Gateway accounts: %s" % method, code=404)104# -----------------105# AUTHORIZERS APIs106# -----------------107def get_authorizer_id_from_path(path):108    match = re.match(PATH_REGEX_AUTHORIZERS, path)109    return match.group(2) if match else None110def _find_authorizer(api_id, authorizer_id):111    return find_api_subentity_by_id(api_id, authorizer_id, "authorizers")112def normalize_authorizer(data):113    is_list = isinstance(data, list)114    entries = data if is_list else [data]115    for i in range(len(entries)):116        entry = common.clone(entries[i])117        # terraform sends this as a string in patch, so convert to int118        entry["authorizerResultTtlInSeconds"] = int(entry.get("authorizerResultTtlInSeconds", 300))119        entries[i] = entry120    return entries if is_list else entries[0]121def get_authorizers(path):122    region_details = APIGatewayRegion.get()123    # This function returns either a list or a single authorizer (depending on the path)124    api_id = get_api_id_from_path(path)125    authorizer_id = get_authorizer_id_from_path(path)126    auth_list = region_details.authorizers.get(api_id) or []127    if authorizer_id:128        authorizer = _find_authorizer(api_id, authorizer_id)129        if authorizer is None:130            return make_error_response(131                "Authorizer not found: %s" % authorizer_id,132                code=404,133                error_type="NotFoundException",134            )135        return to_authorizer_response_json(api_id, authorizer)136    result = [to_authorizer_response_json(api_id, a) for a in auth_list]137    result = {"item": result}138    return result139def add_authorizer(path, data):140    region_details = APIGatewayRegion.get()141    api_id = get_api_id_from_path(path)142    authorizer_id = common.short_uid()[:6]  # length 6 to make TF tests pass143    result = common.clone(data)144    result["id"] = authorizer_id145    result = normalize_authorizer(result)146    region_details.authorizers.setdefault(api_id, []).append(result)147    return make_json_response(to_authorizer_response_json(api_id, result))148def update_authorizer(path, data):149    region_details = APIGatewayRegion.get()150    api_id = get_api_id_from_path(path)151    authorizer_id = get_authorizer_id_from_path(path)152    authorizer = _find_authorizer(api_id, authorizer_id)153    if authorizer is None:154        return make_error_response("Authorizer not found for API: %s" % api_id, code=404)155    result = apply_json_patch_safe(authorizer, data["patchOperations"])156    result = normalize_authorizer(result)157    auth_list = region_details.authorizers[api_id]158    for i in range(len(auth_list)):159        if auth_list[i]["id"] == authorizer_id:160            auth_list[i] = result161    return make_json_response(to_authorizer_response_json(api_id, result))162def delete_authorizer(path):163    region_details = APIGatewayRegion.get()164    api_id = get_api_id_from_path(path)165    authorizer_id = get_authorizer_id_from_path(path)166    auth_list = region_details.authorizers[api_id]167    for i in range(len(auth_list)):168        if auth_list[i]["id"] == authorizer_id:169            del auth_list[i]170            break171    return make_accepted_response()172def handle_authorizers(method, path, data, headers):173    if method == "GET":174        return get_authorizers(path)175    if method == "POST":176        return add_authorizer(path, data)177    if method == "PATCH":178        return update_authorizer(path, data)179    if method == "DELETE":180        return delete_authorizer(path)181    return make_error_response("Not implemented for API Gateway authorizers: %s" % method, code=404)182# -------------------------183# DOCUMENTATION PARTS APIs184# -------------------------185def get_documentation_part_id_from_path(path):186    match = re.match(PATH_REGEX_DOC_PARTS, path)187    return match.group(2) if match else None188def _find_documentation_part(api_id, documentation_part_id):189    return find_api_subentity_by_id(api_id, documentation_part_id, "documentation_parts")190def get_documentation_parts(path):191    region_details = APIGatewayRegion.get()192    # This function returns either a list or a single entity (depending on the path)193    api_id = get_api_id_from_path(path)194    entity_id = get_documentation_part_id_from_path(path)195    auth_list = region_details.documentation_parts.get(api_id) or []196    if entity_id:197        entity = _find_documentation_part(api_id, entity_id)198        if entity is None:199            return make_error_response(200                "Documentation part not found: %s" % entity_id,201                code=404,202                error_type="NotFoundException",203            )204        return to_documentation_part_response_json(api_id, entity)205    result = [to_documentation_part_response_json(api_id, a) for a in auth_list]206    result = {"item": result}207    return result208def add_documentation_part(path, data):209    region_details = APIGatewayRegion.get()210    api_id = get_api_id_from_path(path)211    entity_id = common.short_uid()[:6]  # length 6 to make TF tests pass212    result = common.clone(data)213    result["id"] = entity_id214    region_details.documentation_parts.setdefault(api_id, []).append(result)215    return make_json_response(to_documentation_part_response_json(api_id, result))216def update_documentation_part(path, data):217    region_details = APIGatewayRegion.get()218    api_id = get_api_id_from_path(path)219    entity_id = get_documentation_part_id_from_path(path)220    entity = _find_documentation_part(api_id, entity_id)221    if entity is None:222        return make_error_response("Documentation part not found for API: %s" % api_id, code=404)223    result = apply_json_patch_safe(entity, data["patchOperations"])224    auth_list = region_details.documentation_parts[api_id]225    for i in range(len(auth_list)):226        if auth_list[i]["id"] == entity_id:227            auth_list[i] = result228    return make_json_response(to_documentation_part_response_json(api_id, result))229def delete_documentation_part(path):230    region_details = APIGatewayRegion.get()231    api_id = get_api_id_from_path(path)232    entity_id = get_documentation_part_id_from_path(path)233    auth_list = region_details.documentation_parts[api_id]234    for i in range(len(auth_list)):235        if auth_list[i]["id"] == entity_id:236            del auth_list[i]237            break238    return make_accepted_response()239def handle_documentation_parts(method, path, data, headers):240    if method == "GET":241        return get_documentation_parts(path)242    if method == "POST":243        return add_documentation_part(path, data)244    if method == "PATCH":245        return update_documentation_part(path, data)246    if method == "DELETE":247        return delete_documentation_part(path)248    return make_error_response(249        "Not implemented for API Gateway documentation parts: %s" % method, code=404250    )251# -----------------------252# BASE PATH MAPPING APIs253# -----------------------254def get_domain_from_path(path):255    matched = re.match(PATH_REGEX_PATH_MAPPINGS, path)256    return matched.group(1) if matched else None257def get_base_path_from_path(path):258    return re.match(PATH_REGEX_PATH_MAPPINGS, path).group(2)259def get_base_path_mapping(path):260    region_details = APIGatewayRegion.get()261    # This function returns either a list or a single mapping (depending on the path)262    domain_name = get_domain_from_path(path)263    base_path = get_base_path_from_path(path)264    mappings_list = region_details.base_path_mappings.get(domain_name) or []265    if base_path:266        mapping = ([m for m in mappings_list if m["basePath"] == base_path] or [None])[0]267        if mapping is None:268            return make_error_response(269                "Base path mapping not found: %s" % base_path,270                code=404,271                error_type="NotFoundException",272            )273        return to_base_mapping_response_json(domain_name, base_path, mapping)274    result = [to_base_mapping_response_json(domain_name, m["basePath"], m) for m in mappings_list]275    result = {"item": result}276    return result277def add_base_path_mapping(path, data):278    region_details = APIGatewayRegion.get()279    domain_name = get_domain_from_path(path)280    # Note: "(none)" is a special value in API GW:281    # https://docs.aws.amazon.com/apigateway/api-reference/link-relation/basepathmapping-by-base-path282    base_path = data["basePath"] = data.get("basePath") or "(none)"283    result = common.clone(data)284    region_details.base_path_mappings.setdefault(domain_name, []).append(result)285    return make_json_response(to_base_mapping_response_json(domain_name, base_path, result))286def update_base_path_mapping(path, data):287    region_details = APIGatewayRegion.get()288    domain_name = get_domain_from_path(path)289    base_path = get_base_path_from_path(path)290    mappings_list = region_details.base_path_mappings.get(domain_name) or []291    mapping = ([m for m in mappings_list if m["basePath"] == base_path] or [None])[0]292    if mapping is None:293        return make_error_response(294            "Not found: mapping for domain name %s, base path %s" % (domain_name, base_path),295            code=404,296        )297    operations = data["patchOperations"]298    operations = operations if isinstance(operations, list) else [operations]299    for operation in operations:300        if operation["path"] == "/restapiId":301            operation["path"] = "/restApiId"302    result = apply_json_patch_safe(mapping, operations)303    for i in range(len(mappings_list)):304        if mappings_list[i]["basePath"] == base_path:305            mappings_list[i] = result306    return make_json_response(to_base_mapping_response_json(domain_name, base_path, result))307def delete_base_path_mapping(path):308    region_details = APIGatewayRegion.get()309    domain_name = get_domain_from_path(path)310    base_path = get_base_path_from_path(path)311    mappings_list = region_details.base_path_mappings.get(domain_name) or []312    for i in range(len(mappings_list)):313        if mappings_list[i]["basePath"] == base_path:314            del mappings_list[i]315            return make_accepted_response()316    return make_error_response(317        "Base path mapping %s for domain %s not found" % (base_path, domain_name),318        code=404,319    )320def handle_base_path_mappings(method, path, data, headers):321    if method == "GET":322        return get_base_path_mapping(path)323    if method == "POST":324        return add_base_path_mapping(path, data)325    if method == "PATCH":326        return update_base_path_mapping(path, data)327    if method == "DELETE":328        return delete_base_path_mapping(path)329    return make_error_response(330        "Not implemented for API Gateway base path mappings: %s" % method, code=404331    )332# ------------------------333# CLIENT CERTIFICATE APIs334# ------------------------335def get_cert_id_from_path(path):336    matched = re.match(PATH_REGEX_CLIENT_CERTS, path)337    return matched.group(1) if matched else None338def get_client_certificate(path):339    region_details = APIGatewayRegion.get()340    cert_id = get_cert_id_from_path(path)341    result = region_details.client_certificates.get(cert_id)342    if result is None:343        return make_error_response('Client certificate ID "%s" not found' % cert_id, code=404)344    return result345def add_client_certificate(path, data):346    region_details = APIGatewayRegion.get()347    result = common.clone(data)348    result["clientCertificateId"] = cert_id = common.short_uid()349    result["createdDate"] = common.now_utc()350    result["expirationDate"] = result["createdDate"] + 60 * 60 * 24 * 30  # assume 30 days validity351    result["pemEncodedCertificate"] = "testcert-123"  # TODO return proper certificate!352    region_details.client_certificates[cert_id] = result353    return make_json_response(to_client_cert_response_json(result))354def update_client_certificate(path, data):355    region_details = APIGatewayRegion.get()356    entity_id = get_cert_id_from_path(path)357    entity = region_details.client_certificates.get(entity_id)358    if entity is None:359        return make_error_response('Client certificate ID "%s" not found' % entity_id, code=404)360    result = apply_json_patch_safe(entity, data["patchOperations"])361    return make_json_response(to_client_cert_response_json(result))362def delete_client_certificate(path):363    region_details = APIGatewayRegion.get()364    entity_id = get_cert_id_from_path(path)365    entity = region_details.client_certificates.pop(entity_id, None)366    if entity is None:367        return make_error_response('VPC link ID "%s" not found for deletion' % entity_id, code=404)368    return make_accepted_response()369def handle_client_certificates(method, path, data, headers):370    if method == "GET":371        return get_client_certificate(path)372    if method == "POST":373        return add_client_certificate(path, data)374    if method == "PATCH":375        return update_client_certificate(path, data)376    if method == "DELETE":377        return delete_client_certificate(path)378    return make_error_response(379        "Not implemented for API Gateway base path mappings: %s" % method, code=404380    )381# --------------382# VCP LINK APIs383# --------------384def get_vpc_links(path):385    region_details = APIGatewayRegion.get()386    vpc_link_id = get_vpc_link_id_from_path(path)387    if vpc_link_id:388        vpc_link = region_details.vpc_links.get(vpc_link_id)389        if vpc_link is None:390            return make_error_response('VPC link ID "%s" not found' % vpc_link_id, code=404)391        return make_json_response(to_vpc_link_response_json(vpc_link))392    result = region_details.vpc_links.values()393    result = [to_vpc_link_response_json(r) for r in result]394    result = {"items": result}395    return result396def add_vpc_link(path, data):397    region_details = APIGatewayRegion.get()398    result = common.clone(data)399    result["id"] = common.short_uid()400    result["status"] = "AVAILABLE"401    region_details.vpc_links[result["id"]] = result402    return make_json_response(to_vpc_link_response_json(result))403def update_vpc_link(path, data):404    region_details = APIGatewayRegion.get()405    vpc_link_id = get_vpc_link_id_from_path(path)406    vpc_link = region_details.vpc_links.get(vpc_link_id)407    if vpc_link is None:408        return make_error_response('VPC link ID "%s" not found' % vpc_link_id, code=404)409    result = apply_json_patch_safe(vpc_link, data["patchOperations"])410    return make_json_response(to_vpc_link_response_json(result))411def delete_vpc_link(path):412    region_details = APIGatewayRegion.get()413    vpc_link_id = get_vpc_link_id_from_path(path)414    vpc_link = region_details.vpc_links.pop(vpc_link_id, None)415    if vpc_link is None:416        return make_error_response(417            'VPC link ID "%s" not found for deletion' % vpc_link_id, code=404418        )419    return make_accepted_response()420def get_vpc_link_id_from_path(path):421    match = re.match(PATH_REGEX_VPC_LINKS, path)422    return match.group(1) if match else None423def handle_vpc_links(method, path, data, headers):424    if method == "GET":425        return get_vpc_links(path)426    if method == "POST":427        return add_vpc_link(path, data)428    if method == "PATCH":429        return update_vpc_link(path, data)430    if method == "DELETE":431        return delete_vpc_link(path)432    return make_error_response("Not implemented for API Gateway VPC links: %s" % method, code=404)433# ----------------434# VALIDATORS APIs435# ----------------436def get_validator_id_from_path(path):437    match = re.match(PATH_REGEX_VALIDATORS, path)438    return match.group(2) if match else None439def _find_validator(api_id, validator_id):440    region_details = APIGatewayRegion.get()441    auth_list = region_details.validators.get(api_id) or []442    validator = ([a for a in auth_list if a["id"] == validator_id] or [None])[0]443    return validator444def get_validators(path):445    region_details = APIGatewayRegion.get()446    # This function returns either a list or a single validator (depending on the path)447    api_id = get_api_id_from_path(path)448    validator_id = get_validator_id_from_path(path)449    auth_list = region_details.validators.get(api_id) or []450    if validator_id:451        validator = _find_validator(api_id, validator_id)452        if validator is None:453            return make_error_response(454                "Validator %s for API Gateway %s not found" % (validator_id, api_id),455                code=404,456            )457        return to_validator_response_json(api_id, validator)458    result = [to_validator_response_json(api_id, a) for a in auth_list]459    result = {"item": result}460    return result461def add_validator(path, data):462    region_details = APIGatewayRegion.get()463    api_id = get_api_id_from_path(path)464    validator_id = common.short_uid()[:6]  # length 6 (as in AWS) to make TF tests pass465    result = common.clone(data)466    result["id"] = validator_id467    region_details.validators.setdefault(api_id, []).append(result)468    return result469def update_validator(path, data):470    region_details = APIGatewayRegion.get()471    api_id = get_api_id_from_path(path)472    validator_id = get_validator_id_from_path(path)473    validator = _find_validator(api_id, validator_id)474    if validator is None:475        return make_error_response(476            "Validator %s for API Gateway %s not found" % (validator_id, api_id),477            code=404,478        )479    result = apply_json_patch_safe(validator, data["patchOperations"])480    entry_list = region_details.validators[api_id]481    for i in range(len(entry_list)):482        if entry_list[i]["id"] == validator_id:483            entry_list[i] = result484    return make_json_response(to_validator_response_json(api_id, result))485def delete_validator(path):486    region_details = APIGatewayRegion.get()487    api_id = get_api_id_from_path(path)488    validator_id = get_validator_id_from_path(path)489    auth_list = region_details.validators[api_id]490    for i in range(len(auth_list)):491        if auth_list[i]["id"] == validator_id:492            del auth_list[i]493            return make_accepted_response()494    return make_error_response(495        "Validator %s for API Gateway %s not found" % (validator_id, api_id), code=404496    )497def handle_validators(method, path, data, headers):498    if method == "GET":499        return get_validators(path)500    if method == "POST":501        return add_validator(path, data)502    if method == "PATCH":503        return update_validator(path, data)504    if method == "DELETE":505        return delete_validator(path)506    return make_error_response("Not implemented for API Gateway validators: %s" % method, code=404)507# -----------------------508# GATEWAY RESPONSES APIs509# -----------------------510# TODO: merge with to_response_json(..) above511def gateway_response_to_response_json(item, api_id):512    base_path = "/restapis/%s/gatewayresponses" % api_id513    item["_links"] = {514        "self": {"href": "%s/%s" % (base_path, item["responseType"])},515        "gatewayresponse:put": {516            "href": "%s/{response_type}" % base_path,517            "templated": True,518        },519        "gatewayresponse:update": {"href": "%s/%s" % (base_path, item["responseType"])},520    }521    item["responseParameters"] = item.get("responseParameters", {})522    item["responseTemplates"] = item.get("responseTemplates", {})523    return item524def get_gateway_responses(api_id):525    region_details = APIGatewayRegion.get()526    result = region_details.gateway_responses.get(api_id, [])527    href = "http://docs.aws.amazon.com/apigateway/latest/developerguide/restapi-gatewayresponse-{rel}.html"528    base_path = "/restapis/%s/gatewayresponses" % api_id529    result = {530        "_links": {531            "curies": {"href": href, "name": "gatewayresponse", "templated": True},532            "self": {"href": base_path},533            "first": {"href": base_path},534            "gatewayresponse:by-type": {535                "href": "%s/{response_type}" % base_path,536                "templated": True,537            },538            "item": [{"href": "%s/%s" % (base_path, r["responseType"])} for r in result],539        },540        "_embedded": {"item": [gateway_response_to_response_json(i, api_id) for i in result]},541        # Note: Looks like the format required by aws CLI ("item" at top level) differs from the docs:542        # https://docs.aws.amazon.com/apigateway/api-reference/resource/gateway-responses/543        "item": [gateway_response_to_response_json(i, api_id) for i in result],544    }545    return result546def get_gateway_response(api_id, response_type):547    region_details = APIGatewayRegion.get()548    responses = region_details.gateway_responses.get(api_id, [])549    result = [r for r in responses if r["responseType"] == response_type]550    if result:551        return result[0]552    return make_error_response(553        "Gateway response %s for API Gateway %s not found" % (response_type, api_id),554        code=404,555    )556def put_gateway_response(api_id, response_type, data):557    region_details = APIGatewayRegion.get()558    responses = region_details.gateway_responses.setdefault(api_id, [])559    existing = ([r for r in responses if r["responseType"] == response_type] or [None])[0]560    if existing:561        existing.update(data)562    else:563        data["responseType"] = response_type564        responses.append(data)565    return data566def delete_gateway_response(api_id, response_type):567    region_details = APIGatewayRegion.get()568    responses = region_details.gateway_responses.get(api_id) or []569    region_details.gateway_responses[api_id] = [570        r for r in responses if r["responseType"] != response_type571    ]572    return make_accepted_response()573def update_gateway_response(api_id, response_type, data):574    region_details = APIGatewayRegion.get()575    responses = region_details.gateway_responses.setdefault(api_id, [])576    existing = ([r for r in responses if r["responseType"] == response_type] or [None])[0]577    if existing is None:578        return make_error_response(579            "Gateway response %s for API Gateway %s not found" % (response_type, api_id),580            code=404,581        )582    result = apply_json_patch_safe(existing, data["patchOperations"])583    return result584def handle_gateway_responses(method, path, data, headers):585    search_match = re.search(PATH_REGEX_RESPONSES, path)586    api_id = search_match.group(1)...
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.