Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use cors_error_response method in localstack

Best Python code snippet using localstack_python

slices_controller.py

Source:slices_controller.py

...72        return cors_success_response(response_body=response)73    except OrchestratorException as e:74        logger.exception(e)75        failure_counter.labels(POST_METHOD, SLICES_CREATE_PATH).inc()76        return cors_error_response(error=e)77    except Exception as e:78        logger.exception(e)79        failure_counter.labels(POST_METHOD, SLICES_CREATE_PATH).inc()80        return cors_error_response(error=e)81def slices_delete_slice_id_delete(slice_id) -> Status200OkNoContent:  # noqa: E50182    """Delete slice.83    Request to delete slice. On success, resources associated with slice or sliver are stopped if necessary,84    de-provisioned and un-allocated at the respective sites.  # noqa: E50185    :param slice_id: Slice identified by universally unique identifier86    :type slice_id: str87    :rtype: Status200OkNoContent88    """89    handler = OrchestratorHandler()90    logger = handler.get_logger()91    received_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()92    try:93        token = get_token()94        handler.delete_slice(token=token, slice_id=slice_id)95        success_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()96        slice_info = Status200OkNoContentData()97        slice_info.details = f"Slice '{slice_id}' has been successfully deleted"98        response = Status200OkNoContent()99        response.data = [slice_info]100        response.size = len(response.data)101        response.status = 200102        response.type = 'no_content'103        return cors_success_response(response_body=response)104    except OrchestratorException as e:105        logger.exception(e)106        failure_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()107        return cors_error_response(error=e)108    except Exception as e:109        logger.exception(e)110        failure_counter.labels(DELETE_METHOD, SLICES_DELETE_PATH).inc()111        return cors_error_response(error=e)112def slices_get(name=None, states=None, limit=None, offset=None) -> Slices:  # noqa: E501113    """Retrieve a listing of user slices114    Retrieve a listing of user slices # noqa: E501115    :param name: Search for Slices with the name116    :type name: str117    :param states: Search for Slices in the specified states118    :type states: List[str]119    :param limit: maximum number of results to return per page (1 or more)120    :type limit: int121    :param offset: number of items to skip before starting to collect the result set122    :type offset: int123    :rtype: Slices124    """125    handler = OrchestratorHandler()126    logger = handler.get_logger()127    received_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()128    try:129        token = get_token()130        slices_dict = handler.get_slices(token=token, states=states, name=name, limit=limit, offset=offset)131        response = Slices()132        response.data = []133        response.type = 'slices'134        for s in slices_dict:135            slice_obj = Slice().from_dict(s)136            response.data.append(slice_obj)137        response.size = len(response.data)138        success_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()139        return cors_success_response(response_body=response)140    except OrchestratorException as e:141        logger.exception(e)142        failure_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()143        return cors_error_response(error=e)144    except Exception as e:145        logger.exception(e)146        failure_counter.labels(GET_METHOD, SLICES_GET_PATH).inc()147        return cors_error_response(error=e)148def slices_modify_slice_id_accept_post(slice_id):  # noqa: E501149    """Accept the last modify an existing slice150    Accept the last modify and prune any failed resources from the Slice.151    Also return the accepted slice model back to the user.   # noqa: E501152    :param slice_id: Slice identified by universally unique identifier153    :type slice_id: str154    :rtype: Slivers155    """156    handler = OrchestratorHandler()157    logger = handler.get_logger()158    received_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()159    try:160        token = get_token()161        value = handler.modify_accept(token=token, slice_id=slice_id)162        slice_object = Slice().from_dict(value)163        response = SliceDetails(data=[slice_object], size=1)164        response.type = 'slice_details'165        success_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()166        return cors_success_response(response_body=response)167    except OrchestratorException as e:168        logger.exception(e)169        failure_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()170        return cors_error_response(error=e)171    except Exception as e:172        logger.exception(e)173        failure_counter.labels(POST_METHOD, SLICES_MODIFY_ACCEPT_PATH).inc()174        return cors_error_response(error=e)175def slices_modify_slice_id_put(body, slice_id):  # noqa: E501176    """Modify an existing slice177    Request to modify an existing slice as described in the request. Request would be a graph ML describing the178    experiment topolgy expected after a modify. The supported modify actions include adding or removing nodes,179    components, network services or interfaces of the slice. On success, one or more slivers are allocated,180    containing resources satisfying the request, and assigned to the given slice. This API returns list and181    description of the resources reserved for the slice in the form of Graph ML. Orchestrator would also trigger182    provisioning of these resources asynchronously on the appropriate sites either now or in the future as requested.183    Experimenter can invoke get slice API to get the latest state of the requested resources.   # noqa: E501184    :param body:185    :type body: dict | bytes186    :param slice_id: Slice identified by universally unique identifier187    :type slice_id: str188    :rtype: Slivers189    """190    handler = OrchestratorHandler()191    logger = handler.get_logger()192    received_counter.labels(POST_METHOD, SLICES_MODIFY_PATH).inc()193    try:194        token = get_token()195        slice_graph = body.decode("utf-8")196        slivers_dict = handler.modify_slice(token=token, slice_id=slice_id, slice_graph=slice_graph)197        response = Slivers()198        response.data = []199        for s in slivers_dict:200            sliver = Sliver().from_dict(s)201            response.data.append(sliver)202        response.size = len(response.data)203        response.type = "slivers"204        success_counter.labels(POST_METHOD, SLICES_MODIFY_PATH).inc()205        return cors_success_response(response_body=response)206    except OrchestratorException as e:207        logger.exception(e)208        failure_counter.labels(POST_METHOD, SLICES_MODIFY_PATH).inc()209        return cors_error_response(error=e)210    except Exception as e:211        logger.exception(e)212        failure_counter.labels(POST_METHOD, SLICES_CREATE_PATH).inc()213        return cors_error_response(error=e)214def slices_renew_slice_id_post(slice_id, lease_end_time) -> Status200OkNoContent:  # noqa: E501215    """Renew slice216    Request to extend slice be renewed with their expiration extended. If possible, the orchestrator should extend the217    slivers to the requested expiration time, or to a sooner time if policy limits apply.  # noqa: E501218    :param slice_id: Slice identified by universally unique identifier219    :type slice_id: str220    :param lease_end_time: New Lease End Time for the Slice221    :type lease_end_time: str222    :rtype: Status200OkNoContent223    """224    handler = OrchestratorHandler()225    logger = handler.get_logger()226    received_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()227    try:228        token = get_token()229        handler.renew_slice(token=token, slice_id=slice_id, new_lease_end_time=lease_end_time)230        success_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()231        slice_info = Status200OkNoContentData()232        slice_info.details = f"Slice '{slice_id}' has been successfully renewed"233        response = Status200OkNoContent()234        response.data = [slice_info]235        response.size = len(response.data)236        response.status = 200237        response.type = 'no_content'238        return cors_success_response(response_body=response)239    except OrchestratorException as e:240        logger.exception(e)241        failure_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()242        return cors_error_response(error=e)243    except Exception as e:244        logger.exception(e)245        failure_counter.labels(POST_METHOD, SLICES_RENEW_PATH).inc()246        return cors_error_response(error=e)247def slices_slice_id_get(slice_id, graph_format) -> SliceDetails:  # noqa: E501248    """slice properties249    Retrieve Slice properties # noqa: E501250    :param slice_id: Slice identified by universally unique identifier251    :type slice_id: str252    :param graph_format: graph format253    :type graph_format: str254    :rtype: SliceDetails255    """256    handler = OrchestratorHandler()257    logger = handler.get_logger()258    received_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()259    try:260        token = get_token()261        value = handler.get_slice_graph(token=token, slice_id=slice_id, graph_format_str=graph_format)262        slice_object = Slice().from_dict(value)263        response = SliceDetails(data=[slice_object], size=1)264        response.type = 'slice_details'265        success_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()266        return cors_success_response(response_body=response)267    except OrchestratorException as e:268        logger.exception(e)269        failure_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()270        return cors_error_response(error=e)271    except Exception as e:272        logger.exception(e)273        failure_counter.labels(GET_METHOD, SLICES_GET_SLICE_ID_PATH).inc()...

slivers_controller.py

Source:slivers_controller.py

...55        return cors_success_response(response_body=response)56    except OrchestratorException as e:57        logger.exception(e)58        failure_counter.labels(GET_METHOD, SLIVERS_GET_PATH).inc()59        return cors_error_response(error=e)60    except Exception as e:61        logger.exception(e)62        failure_counter.labels(GET_METHOD, SLIVERS_GET_PATH).inc()63        return cors_error_response(error=e)64def slivers_sliver_id_get(slice_id, sliver_id) -> Slivers:  # noqa: E50165    """slivers properties66    Retrieve Sliver properties # noqa: E50167    :param slice_id: Slice identified by universally unique identifier68    :type slice_id: str69    :param sliver_id: Sliver identified by universally unique identifier70    :type sliver_id: str71    :rtype: Slivers72    """73    handler = OrchestratorHandler()74    logger = handler.get_logger()75    received_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()76    try:77        token = get_token()78        slivers_dict = handler.get_slivers(slice_id=slice_id, token=token, sliver_id=sliver_id)79        response = Slivers()80        response.data = []81        for s in slivers_dict:82            sliver = Sliver().from_dict(s)83            response.data.append(sliver)84        response.size = len(response.data)85        response.type = "slivers"86        success_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()87        return cors_success_response(response_body=response)88    except OrchestratorException as e:89        logger.exception(e)90        failure_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()91        return cors_error_response(error=e)92    except Exception as e:93        logger.exception(e)94        failure_counter.labels(GET_METHOD, SLIVERS_GET_SLIVER_ID_PATH).inc()...

resources_controller.py

Source:resources_controller.py

...50        return cors_success_response(response_body=response)51    except OrchestratorException as e:52        logger.exception(e)53        failure_counter.labels(GET_METHOD, PORTAL_RESOURCES_PATH).inc()54        return cors_error_response(error=e)55    except Exception as e:56        logger.exception(e)57        failure_counter.labels(GET_METHOD, PORTAL_RESOURCES_PATH).inc()58        return cors_error_response(error=e)59def resources_get(level, force_refresh) -> Resources:  # noqa: E50160    """Retrieve a listing and description of available resources61    Retrieve a listing and description of available resources # noqa: E50162    :param level: Level of details63    :type level: int64    :param force_refresh: Force to retrieve current available resource information.65    :type force_refresh: bool66    :rtype: Resources67    """68    handler = OrchestratorHandler()69    logger = handler.get_logger()70    received_counter.labels(GET_METHOD, RESOURCES_PATH).inc()71    try:72        token = get_token()73        bqm_dict = handler.list_resources(token=token, level=level, force_refresh=force_refresh)74        response = Resources()75        response.data = [Resource().from_dict(bqm_dict)]76        response.size = 177        response.type = "resources"78        success_counter.labels(GET_METHOD, RESOURCES_PATH).inc()79        return cors_success_response(response_body=response)80    except OrchestratorException as e:81        logger.exception(e)82        failure_counter.labels(GET_METHOD, RESOURCES_PATH).inc()83        return cors_error_response(error=e)84    except Exception as e:85        logger.exception(e)86        failure_counter.labels(GET_METHOD, RESOURCES_PATH).inc()...

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.