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

How to use custom_path_addressing_rules method in localstack

Best Python code snippet using localstack_python

service_router.py

Source:service_router.py

...103    Rules based on the host header of the request.104    """105    if ".execute-api." in host:106        return "apigateway"107def custom_path_addressing_rules(path: str) -> Optional[str]:108    """109    Rules which are only based on the request path.110    """111    if is_sqs_queue_url(path):112        return "sqs"113    if path.startswith("/2015-03-31/functions/"):114        return "lambda"115def legacy_rules(request: Request) -> Optional[str]:116    """117    *Legacy* rules which migrate routing logic which will become obsolete with the ASF Gateway.118    All rules which are implemented here should be migrated to the new router once these services are migrated to ASF.119    TODO: These custom rules should become obsolete by migrating these to use the http/router.py120    """121    path = request.path122    method = request.method123    host = hostname_from_url(request.host)124    # API Gateway invocation URLs125    if ("/%s/" % PATH_USER_REQUEST) in request.path or (126        host.endswith(LOCALHOST_HOSTNAME) and "execute-api" in host127    ):128        return "apigateway"129    # DynamoDB shell URLs130    if path.startswith("/shell") or path.startswith("/dynamodb/shell"):131        return "dynamodb"132    # TODO The remaining rules here are special S3 rules - needs to be discussed how these should be handled.133    #      Some are similar to other rules and not that greedy, others are nearly general fallbacks.134    stripped = path.strip("/")135    if method in ["GET", "HEAD"] and stripped:136        # assume that this is an S3 GET request with URL path `/<bucket>/<key ...>`137        return "s3"138    # detect S3 URLs139    if stripped and "/" not in stripped:140        if method == "PUT":141            # assume that this is an S3 PUT bucket request with URL path `/<bucket>`142            return "s3"143        if method == "POST" and "key" in request.values:144            # assume that this is an S3 POST request with form parameters or multipart form in the body145            return "s3"146    # detect S3 requests sent from aws-cli using --no-sign-request option147    if "aws-cli/" in str(request.user_agent):148        return "s3"149    # detect S3 pre-signed URLs150    values = request.values151    if "AWSAccessKeyId" in values or "Signature" in values:152        return "s3"153    # S3 delete object requests154    if method == "POST" and "delete" in values:155        data_bytes = to_bytes(request.get_data())156        if b"<Delete" in data_bytes and b"<Key>" in data_bytes:157            return "s3"158    # Put Object API can have multiple keys159    if stripped.count("/") >= 1 and method == "PUT":160        # assume that this is an S3 PUT bucket object request with URL path `/<bucket>/object`161        # or `/<bucket>/object/object1/+`162        return "s3"163    # detect S3 requests with "AWS id:key" Auth headers164    auth_header = request.headers.get("Authorization") or ""165    if auth_header.startswith("AWS "):166        return "s3"167    if uses_host_addressing(request.headers):168        # Note: This needs to be the last rule (and therefore is not in the host rules), since it is incredibly greedy169        return "s3"170@lru_cache()171def get_service_catalog() -> ServiceCatalog:172    """Loads the ServiceCatalog (which contains all the service specs)."""173    return ServiceCatalog()174def determine_aws_service_name(175    request: Request, services: ServiceCatalog = get_service_catalog()176) -> Optional[str]:177    """178    Tries to determine the name of the AWS service an incoming request is targeting.179    :param request: to determine the target service name of180    :param services: service catalog (can be handed in for caching purposes)181    :return: service name string (or None if the targeting service could not be determined exactly)182    """183    signing_name, target_prefix, operation, host, path = _extract_service_indicators(request)184    candidates = set()185    # 1. check the signing names186    if signing_name:187        signing_name_candidates = services.by_signing_name(signing_name)188        if len(signing_name_candidates) == 1:189            # a unique signing-name -> service name mapping is the case for ~75% of service operations190            return signing_name_candidates[0]191        # try to find a match with the custom signing name rules192        custom_match = custom_signing_name_rules(signing_name, path)193        if custom_match:194            return custom_match195        # still ambiguous - add the services to the list of candidates196        candidates.update(signing_name_candidates)197    # 2. check the target prefix198    if target_prefix and operation:199        target_candidates = services.by_target_prefix(target_prefix)200        if len(target_candidates) == 1:201            # a unique target prefix202            return target_candidates[0]203        # still ambiguous - add the services to the list of candidates204        candidates.update(target_candidates)205        # exclude services where the operation is not contained in the service spec206        for service_name in list(candidates):207            service = services.get(service_name)208            if operation not in service.operation_names:209                candidates.remove(service_name)210    else:211        # exclude services which have a target prefix (the current request does not have one)212        for service_name in list(candidates):213            service = services.get(service_name)214            if service.metadata.get("targetPrefix") is not None:215                candidates.remove(service_name)216    if len(candidates) == 1:217        return candidates.pop()218    # 3. check the path219    if path:220        # iterate over the service spec's endpoint prefix221        for prefix, services_per_prefix in services.endpoint_prefix_index.items():222            if path.startswith(prefix):223                if len(services_per_prefix) == 1:224                    return services_per_prefix[0]225                candidates.update(services_per_prefix)226        # try to find a match with the custom path rules227        custom_path_match = custom_path_addressing_rules(path)228        if custom_path_match:229            return custom_path_match230    # 4. check the host (custom host addressing rules)231    if host:232        custom_host_match = custom_host_addressing_rules(host)233        if custom_host_match:234            return custom_host_match235    # 5. check the query / form-data236    values = request.values237    if "Action" in values and "Version" in values:238        # query / ec2 protocol requests always have an action and a version (the action is more significant)239        query_candidates = services.by_operation(values["Action"])240        if len(query_candidates) == 1:241            return query_candidates[0]...

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.