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

How to use is_trace_logging_enabled method in localstack

Best Python code snippet using localstack_python

edge.py

Source:edge.py

...84        port = None85        if api:86            port = get_service_port_for_account(api, headers)87        set_default_region_in_headers(headers)88        should_log_trace = is_trace_logging_enabled(headers)89        if api and should_log_trace:90            # print request trace for debugging, if enabled91            LOG.debug(92                'IN(%s): "%s %s" - headers: %s - data: %s', api, method, path, dict(headers), data93            )94        if not port:95            if method == "OPTIONS":96                if api and should_log_trace:97                    # print request trace for debugging, if enabled98                    LOG.debug('IN(%s): "%s %s" - status: %s', api, method, path, 200)99                return 200100            if api in ["", None, API_UNKNOWN]:101                truncated = truncate(data)102                if auth_header or target or data or path not in ["/", "/favicon.ico"]:103                    LOG.info(104                        (105                            'Unable to find forwarding rule for host "%s", path "%s %s", '106                            'target header "%s", auth header "%s", data "%s"'107                        ),108                        host,109                        method,110                        path,111                        target,112                        auth_header,113                        truncated,114                    )115            else:116                LOG.info(117                    (118                        'Unable to determine forwarding port for API "%s" - please '119                        "make sure this API is enabled via the SERVICES configuration"120                    ),121                    api,122                )123            response = Response()124            response.status_code = 404125            response._content = '{"status": "running"}'126            return response127        if api and not headers.get("Authorization"):128            headers["Authorization"] = aws_stack.mock_aws_request_headers(api)["Authorization"]129        headers[HEADER_TARGET_API] = str(api)130        headers["Host"] = host131        if isinstance(data, dict):132            data = json.dumps(data)133        encoding_type = headers.get("Content-Encoding") or ""134        if encoding_type.upper() == GZIP_ENCODING.upper() and api not in SKIP_GZIP_APIS:135            headers.set("Content-Encoding", IDENTITY_ENCODING)136            data = gzip.decompress(data)137        is_internal_call = is_internal_call_context(headers)138        self._require_service(api)139        lock_ctx = BOOTSTRAP_LOCK140        if is_internal_call or persistence.is_persistence_restored():141            lock_ctx = empty_context_manager()142        with lock_ctx:143            result = do_forward_request(api, method, path, data, headers, port=port)144            if should_log_trace and result not in [None, False, True]:145                result_status_code = getattr(result, "status_code", result)146                result_headers = getattr(result, "headers", {})147                result_content = getattr(result, "content", "")148                LOG.debug(149                    'OUT(%s): "%s %s" - status: %s - response headers: %s - response: %s',150                    api,151                    method,152                    path,153                    result_status_code,154                    dict(result_headers or {}),155                    result_content,156                )157            return result158    def return_response(self, method, path, data, headers, response):159        api = headers.get(HEADER_TARGET_API) or ""160        if is_trace_logging_enabled(headers):161            # print response trace for debugging, if enabled162            if api and api != API_UNKNOWN:163                LOG.debug(164                    'OUT(%s): "%s %s" - status: %s - response headers: %s - response: %s',165                    api,166                    method,167                    path,168                    response.status_code,169                    dict(response.headers),170                    response.content,171                )172        if (173            response._content174            and headers.get("Accept-Encoding") == "gzip"175            and api not in SKIP_GZIP_APIS176            and not response.headers.pop(HEADER_SKIP_RESPONSE_ZIPPING, None)177        ):178            # services may decide to set HEADER_SKIP_RESPONSE_ZIPPING in the response, to skip result transformations179            response._content = gzip.compress(to_bytes(response._content))180            response.headers["Content-Length"] = str(len(response._content))181            response.headers["Content-Encoding"] = "gzip"182    def _require_service(self, api):183        if not self.service_manager.exists(api):184            raise HTTPErrorResponse("no provider exists for service %s" % api, code=500)185        try:186            self.service_manager.require(api)187        except Exception as e:188            raise HTTPErrorResponse("failed to get service for %s: %s" % (api, e), code=500)189def do_forward_request(api, method, path, data, headers, port=None):190    if config.FORWARD_EDGE_INMEM:191        result = do_forward_request_inmem(api, method, path, data, headers, port=port)192    else:193        result = do_forward_request_network(port, method, path, data, headers)194    if hasattr(result, "status_code") and int(result.status_code) >= 400 and method == "OPTIONS":195        # fall back to successful response for OPTIONS requests196        return 200197    return result198def get_handler_for_api(api, headers):199    return PROXY_LISTENERS.get(api)200def do_forward_request_inmem(api, method, path, data, headers, port=None):201    listener_details = get_handler_for_api(api, headers)202    if not listener_details:203        message = f'Unable to find listener for service "{api}" - please make sure to include it in $SERVICES'204        LOG.warning(message)205        raise HTTPErrorResponse(message, code=400)206    service_name, backend_port, listener = listener_details207    # TODO determine client address..?208    client_address = LOCALHOST_IP209    server_address = headers.get("host") or LOCALHOST210    forward_url = f"http://{LOCALHOST}:{backend_port}"211    response = modify_and_forward(212        method=method,213        path=path,214        data_bytes=data,215        headers=headers,216        forward_base_url=forward_url,217        listeners=[listener],218        client_address=client_address,219        server_address=server_address,220    )221    return response222def do_forward_request_network(port, method, path, data, headers, target_url=None):223    # TODO: enable per-service endpoints, to allow deploying in distributed settings224    target_url = target_url or f"{config.get_protocol()}://{LOCALHOST}:{port}"225    url = f"{target_url}{path}"226    return requests.request(227        method,228        url,229        data=data,230        headers=headers,231        verify=False,232        stream=True,233        allow_redirects=False,234    )235def get_auth_string(method, path, headers, data=None):236    """237    Get Auth header from Header (this is how aws client's like boto typically238    provide it) or from query string or url encoded parameters sometimes239    happens with presigned requests. Always return to the Authorization Header240    form.241    Typically, an auth string comes in as a header:242        Authorization: AWS4-HMAC-SHA256 \243        Credential=_not_needed_locally_/20210312/us-east-1/sqs/aws4_request, \244        SignedHeaders=content-type;host;x-amz-date, \245        Signature=9277c941f4ecafcc0f290728e50cd7a3fa0e41763fbd2373fcdd3faf2dbddc2e246    Here's what Authorization looks like as part of an presigned GET request:247       &X-Amz-Algorithm=AWS4-HMAC-SHA256\248       &X-Amz-Credential=test%2F20210313%2Fus-east-1%2Fsqs%2Faws4_request\249       &X-Amz-Date=20210313T011059Z&X-Amz-Expires=86400000&X-Amz-SignedHeaders=host\250       &X-Amz-Signature=2c652c7bc9a3b75579db3d987d1e6dd056f0ac776c1e1d4ec91e2ce84e5ad3ae251    """252    if auth_header := headers.get("authorization", ""):253        return auth_header254    data_components = parse_request_data(method, path, data)255    algorithm = data_components.get("X-Amz-Algorithm")256    credential = data_components.get("X-Amz-Credential")257    signature = data_components.get("X-Amz-Signature")258    signed_headers = data_components.get("X-Amz-SignedHeaders")259    if algorithm and credential and signature and signed_headers:260        return (261            f"{algorithm} Credential={credential}, "262            + f"SignedHeaders={signed_headers}, "263            + f"Signature={signature}"264        )265    return ""266def get_service_port_for_account(service, headers):267    # assume we're only using a single account, hence return the static port mapping from config.py268    return config.service_port(service)269PROXY_LISTENER_EDGE = ProxyListenerEdge()270# the ROUTER is part of the edge proxy. use the router to inject custom handlers that are handled before actual271# service calls272ROUTER: Router[Handler] = Router(273    dispatcher=handler_dispatcher(), converters={"regex": RegexConverter}274)275def is_trace_logging_enabled(headers):276    if not config.LS_LOG:277        return False278    if config.LS_LOG == LS_LOG_TRACE_INTERNAL:279        return True280    auth_header = headers.get("Authorization") or ""281    return INTERNAL_AWS_ACCESS_KEY_ID not in auth_header282def do_start_edge(bind_address, port, use_ssl, asynchronous=False):283    from localstack.http.adapters import RouterListener284    from localstack.services.internal import LocalstackResourceHandler285    start_dns_server(asynchronous=True)286    listeners = [287        LocalstackResourceHandler(),  # handle internal resources first288        RouterListener(ROUTER),  # then custom routes289        PROXY_LISTENER_EDGE,  # then call the edge proxy listener...

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.