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

How to use remove_xml_preamble method in localstack

Best Python code snippet using localstack_python

s3_listener.py

Source:s3_listener.py

...529    if "LocationConstraint" in content:530        pattern = r"<LocationConstraint([^>]*)>\s*</LocationConstraint>"531        replace = r"<LocationConstraint\1>%s</LocationConstraint>" % aws_stack.get_region()532        response._content = re.sub(pattern, replace, content)533        remove_xml_preamble(response)534def fix_range_content_type(bucket_name, path, headers, response):535    # Fix content type for Range requests - https://github.com/localstack/localstack/issues/1259536    if "Range" not in headers:537        return538    if response.status_code >= 400:539        return540    s3_client = aws_stack.connect_to_service("s3")541    path = urlparse.urlparse(urlparse.unquote(path)).path542    key_name = extract_key_name(headers, path)543    result = s3_client.head_object(Bucket=bucket_name, Key=key_name)544    content_type = result["ContentType"]545    if response.headers.get("Content-Type") == "text/html; charset=utf-8":546        response.headers["Content-Type"] = content_type547def fix_delete_objects_response(bucket_name, method, parsed_path, data, headers, response):548    # Deleting non-existing keys should not result in errors.549    # Fixes https://github.com/localstack/localstack/issues/1893550    if not (method == "POST" and parsed_path.query == "delete" and "<Delete" in to_str(data or "")):551        return552    content = to_str(response._content)553    if "<Error>" not in content:554        return555    result = xmltodict.parse(content).get("DeleteResult")556    errors = result.get("Error")557    errors = errors if isinstance(errors, list) else [errors]558    deleted = result.get("Deleted")559    if not isinstance(result.get("Deleted"), list):560        deleted = result["Deleted"] = [deleted] if deleted else []561    for entry in list(errors):562        if set(entry.keys()) == set(["Key"]):563            errors.remove(entry)564            deleted.append(entry)565    if not errors:566        result.pop("Error")567    response._content = xmltodict.unparse({"DeleteResult": result})568def fix_metadata_key_underscores(request_headers={}, response=None):569    # fix for https://github.com/localstack/localstack/issues/1790570    underscore_replacement = "---"571    meta_header_prefix = "x-amz-meta-"572    prefix_len = len(meta_header_prefix)573    updated = False574    for key in list(request_headers.keys()):575        if key.lower().startswith(meta_header_prefix):576            key_new = meta_header_prefix + key[prefix_len:].replace("_", underscore_replacement)577            if key != key_new:578                request_headers[key_new] = request_headers.pop(key)579                updated = True580    if response is not None:581        for key in list(response.headers.keys()):582            if key.lower().startswith(meta_header_prefix):583                key_new = meta_header_prefix + key[prefix_len:].replace(underscore_replacement, "_")584                if key != key_new:585                    response.headers[key_new] = response.headers.pop(key)586    return updated587def fix_creation_date(method, path, response):588    if method != "GET" or path != "/":589        return590    response._content = re.sub(591        r"(\.[0-9]+)(\+00:00)?</CreationDate>",592        r"\1Z</CreationDate>",593        to_str(response._content),594    )595def fix_delimiter(data, headers, response):596    if response.status_code == 200 and response._content:597        c, xml_prefix, delimiter = response._content, "<?xml", "<Delimiter><"598        pattern = "[<]Delimiter[>]None[<]"599        if isinstance(c, bytes):600            xml_prefix, delimiter = xml_prefix.encode(), delimiter.encode()601            pattern = pattern.encode()602        if c.startswith(xml_prefix):603            response._content = re.compile(pattern).sub(delimiter, c)604def convert_to_chunked_encoding(method, path, response):605    if method != "GET" or path != "/":606        return607    if response.headers.get("Transfer-Encoding", "").lower() == "chunked":608        return609    response.headers["Transfer-Encoding"] = "chunked"610    response.headers.pop("Content-Encoding", None)611    response.headers.pop("Content-Length", None)612def unquote(s):613    if (s[0], s[-1]) in (('"', '"'), ("'", "'")):614        return s[1:-1]615    return s616def ret304_on_etag(data, headers, response):617    etag = response.headers.get("ETag")618    if etag:619        match = headers.get("If-None-Match")620        if match and unquote(match) == unquote(etag):621            response.status_code = 304622            response._content = ""623def fix_etag_for_multipart(data, headers, response):624    # Fix for https://github.com/localstack/localstack/issues/1978625    if headers.get(CONTENT_SHA256_HEADER) == STREAMING_HMAC_PAYLOAD:626        try:627            if b"chunk-signature=" not in to_bytes(data):628                return629            correct_hash = md5(strip_chunk_signatures(data))630            tags = r"<ETag>%s</ETag>"631            pattern = r"(&#34;)?([^<&]+)(&#34;)?"632            replacement = r"\g<1>%s\g<3>" % correct_hash633            response._content = re.sub(tags % pattern, tags % replacement, to_str(response.content))634            if response.headers.get("ETag"):635                response.headers["ETag"] = re.sub(pattern, replacement, response.headers["ETag"])636        except Exception:637            pass638def remove_xml_preamble(response):639    """Removes <?xml ... ?> from a response content"""640    response._content = re.sub(r"^<\?[^\?]+\?>", "", to_str(response._content))641# --------------642# HELPER METHODS643#   for lifecycle/replication/...644# --------------645def get_lifecycle(bucket_name):646    bucket_name = normalize_bucket_name(bucket_name)647    exists, code, body = is_bucket_available(bucket_name)648    if not exists:649        return xml_response(body, status_code=code)650    lifecycle = BUCKET_LIFECYCLE.get(bucket_name)651    status_code = 200652    if not lifecycle:...

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.