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

How to use _post_process_response_headers method in localstack

Best Python code snippet using localstack_python

generic_proxy.py

Source:generic_proxy.py

...238            status_code=response.status_code,239            content=self._adjust_partition(response.content),240            headers=self._adjust_partition(response.headers),241        )242        self._post_process_response_headers(response)243        return response244    def _adjust_partition_in_path(self, path: str, static_partition: str = None):245        """Adjusts the (still url encoded) URL path"""246        parsed_url = urlparse(path)247        # Make sure to keep blank values, otherwise we drop query params which do not have a248        # value (f.e. "/?policy")249        decoded_query = parse_qs(qs=parsed_url.query, keep_blank_values=True)250        adjusted_path = self._adjust_partition(parsed_url.path, static_partition)251        adjusted_query = self._adjust_partition(decoded_query, static_partition)252        encoded_query = urlencode(adjusted_query, doseq=True)253        # Make sure to avoid empty equals signs (in between and in the end)254        encoded_query = encoded_query.replace("=&", "&")255        encoded_query = re.sub(r"=$", "", encoded_query)256        return f"{adjusted_path}{('?' + encoded_query) if encoded_query else ''}"257    def _adjust_partition(self, source, static_partition: str = None):258        # Call this function recursively if we get a dictionary or a list259        if isinstance(source, dict):260            result = {}261            for k, v in source.items():262                result[k] = self._adjust_partition(v, static_partition)263            return result264        if isinstance(source, list):265            result = []266            for v in source:267                result.append(self._adjust_partition(v, static_partition))268            return result269        elif isinstance(source, bytes):270            try:271                decoded = unquote(to_str(source))272                adjusted = self._adjust_partition(decoded, static_partition)273                return to_bytes(adjusted)274            except UnicodeDecodeError:275                # If the body can't be decoded to a string, we return the initial source276                return source277        elif not isinstance(source, str):278            # Ignore any other types279            return source280        return self.arn_regex.sub(lambda m: self._adjust_match(m, static_partition), source)281    def _adjust_match(self, match: Match, static_partition: str = None):282        region = match.group("Region")283        partition = self._partition_lookup(region) if static_partition is None else static_partition284        service = match.group("Service")285        account_id = match.group("AccountID")286        resource_path = match.group("ResourcePath")287        return f"arn:{partition}:{service}:{region}:{account_id}:{resource_path}"288    def _partition_lookup(self, region: str):289        try:290            partition = self._get_partition_for_region(region)291        except ArnPartitionRewriteListener.InvalidRegionException:292            try:293                # If the region is not properly set (f.e. because it is set to a wildcard),294                # the partition is determined based on the default region.295                partition = self._get_partition_for_region(config.DEFAULT_REGION)296            except self.InvalidRegionException:297                # If it also fails with the DEFAULT_REGION, we use us-east-1 as a fallback298                partition = self._get_partition_for_region(AWS_REGION_US_EAST_1)299        return partition300    @staticmethod301    def _get_partition_for_region(region: Optional[str]) -> str:302        # Region-Partition matching is based on the "regionRegex" definitions in the endpoints.json303        # in the botocore package.304        if region and region.startswith("us-gov-"):305            return "aws-us-gov"306        elif region and region.startswith("us-iso-"):307            return "aws-iso"308        elif region and region.startswith("us-isob-"):309            return "aws-iso-b"310        elif region and region.startswith("cn-"):311            return "aws-cn"312        elif region and re.match(r"^(us|eu|ap|sa|ca|me|af)-\w+-\d+$", region):313            return "aws"314        else:315            raise ArnPartitionRewriteListener.InvalidRegionException(316                f"Region ({region}) could not be matched to a partition."317            )318    @staticmethod319    def _post_process_response_headers(response: RoutingResponse) -> None:320        """Adjust potential content lengths and checksums after modifying the response."""321        if response.headers and response.content:322            if "Content-Length" in response.headers:323                response.headers["Content-Length"] = str(len(to_bytes(response.content)))324            if "x-amz-crc32" in response.headers:325                response.headers["x-amz-crc32"] = calculate_crc32(response.content)326# -------------------327# BASE BACKEND UTILS328# -------------------329T = TypeVar("T", bound="RegionBackend")330class RegionBackend:331    """Base class for provider backends. RegionBackend lookup methods are not thread safe.332    RegionBackend transparently performs namespacing per AWS account IDs.333    Usage...

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.