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

How to use do_forward_request_network method in localstack

Best Python code snippet using localstack_python

edge.py

Source:edge.py

...64        headers.get(HEADER_KILL_SIGNAL) and sys.exit(0)65        if events.infra_stopping.is_set():66            return 50367        if config.EDGE_FORWARD_URL:68            return do_forward_request_network(69                0, method, path, data, headers, target_url=config.EDGE_FORWARD_URL70            )71        target = headers.get("x-amz-target", "")72        auth_header = get_auth_string(method, path, headers, data)73        if auth_header and not headers.get("authorization"):74            headers["authorization"] = auth_header75        host = headers.get("host", "")76        orig_req_url = headers.pop(HEADER_LOCALSTACK_REQUEST_URL, "")77        headers[HEADER_LOCALSTACK_EDGE_URL] = (78            re.sub(r"^([^:]+://[^/]+).*", r"\1", orig_req_url) or "http://%s" % host79        )80        # re-create an HTTP request from the given parts81        request = create_request_from_parts(method, path, data, headers)82        from localstack.aws.protocol.service_router import determine_aws_service_name83        api = determine_aws_service_name(request)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    """...

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.