testbench/proto2rest.py
def __postprocess_object_rest(metadata):
    """The protos for storage/v2 renamed some fields in ways that require some custom coding."""
    # For some fields the storage/v2 name just needs to change slightly.
    bucket_id = testbench.common.bucket_name_from_proto(metadata.get("bucket", None))
    metadata = testbench.common.rest_adjust(
        metadata,
        {
            "bucket": lambda x: ("bucket", bucket_id),
            "createTime": lambda x: ("timeCreated", x),
            "updateTime": lambda x: ("updated", x),
            "kmsKey": lambda x: ("kmsKeyName", x),
            "retentionExpireTime": lambda x: ("retentionExpirationTime", x),
            "deleteTime": lambda x: ("timeDeleted", x),
            "updateStorageClassTime": lambda x: ("timeStorageClassUpdated", x),
            "customerEncryption": lambda x: (
                "customerEncryption",
                __postprocess_customer_encryption(x),
            ),
        },
    )
    metadata["kind"] = "storage#object"
    metadata["id"] = "%s/o/%s/%s" % (
        metadata["bucket"],
        metadata["name"],
        metadata["generation"],
    )
    # Checksums need special treatment
    cs = metadata.pop("checksums", None)
    if cs is not None:
        if "crc32c" in cs:
            metadata["crc32c"] = base64.b64encode(
                struct.pack(">I", cs["crc32c"])
            ).decode("utf-8")
        if "md5Hash" in cs:
            metadata["md5Hash"] = cs["md5Hash"]
    # Finally the ACLs, if present, require additional fields
    if "acl" in metadata:
        metadata["acl"] = [
            __postprocess_object_access_control(
                bucket_id, metadata["name"], metadata["generation"], a
            )
            for a in metadata["acl"]
        ]
    return metadata
The protos for storage/v2 renamed some fields in ways that require some custom coding.
The protos for storage/v2 renamed some fields in ways that require some custom coding.
The protos for storage / v2 renamed some fields in ways that require some custom coding .
def __postprocess_object_rest(metadata):
    bucket_id = testbench.common.bucket_name_from_proto(metadata.get("bucket", None))
    metadata = testbench.common.rest_adjust(
        metadata,
        {
            "bucket": lambda x: ("bucket", bucket_id),
            "createTime": lambda x: ("timeCreated", x),
            "updateTime": lambda x: ("updated", x),
            "kmsKey": lambda x: ("kmsKeyName", x),
            "retentionExpireTime": lambda x: ("retentionExpirationTime", x),
            "deleteTime": lambda x: ("timeDeleted", x),
            "updateStorageClassTime": lambda x: ("timeStorageClassUpdated", x),
            "customerEncryption": lambda x: (
                "customerEncryption",
                __postprocess_customer_encryption(x),
            ),
        },
    )
    metadata["kind"] = "storage#object"
    metadata["id"] = "%s/o/%s/%s" % (
        metadata["bucket"],
        metadata["name"],
        metadata["generation"],
    )
    cs = metadata.pop("checksums", None)
    if cs is not None:
        if "crc32c" in cs:
            metadata["crc32c"] = base64.b64encode(
                struct.pack(">I", cs["crc32c"])
            ).decode("utf-8")
        if "md5Hash" in cs:
            metadata["md5Hash"] = cs["md5Hash"]
    if "acl" in metadata:
        metadata["acl"] = [
            __postprocess_object_access_control(
                bucket_id, metadata["name"], metadata["generation"], a
            )
            for a in metadata["acl"]
        ]
    return metadata
def __postprocess_object_rest ( metadata ) : bucket_id = testbench . common . bucket_name_from_proto ( metadata . get ( "bucket" , None ) ) metadata = testbench . common . rest_adjust ( metadata , { "bucket" : lambda x : ( "bucket" , bucket_id ) , "createTime" : lambda x : ( "timeCreated" , x ) , "updateTime" : lambda x : ( "updated" , x ) , "kmsKey" : lambda x : ( "kmsKeyName" , x ) , "retentionExpireTime" : lambda x : ( "retentionExpirationTime" , x ) , "deleteTime" : lambda x : ( "timeDeleted" , x ) , "updateStorageClassTime" : lambda x : ( "timeStorageClassUpdated" , x ) , "customerEncryption" : lambda x : ( "customerEncryption" , __postprocess_customer_encryption ( x ) , ) , } , ) metadata [ "kind" ] = "storage#object" metadata [ "id" ] = "%s/o/%s/%s" % ( metadata [ "bucket" ] , metadata [ "name" ] , metadata [ "generation" ] , ) cs = metadata . pop ( "checksums" , None ) if cs is not None : if "crc32c" in cs : metadata [ "crc32c" ] = base64 . b64encode ( struct . pack ( ">I" , cs [ "crc32c" ] ) ) . decode ( "utf-8" ) if "md5Hash" in cs : metadata [ "md5Hash" ] = cs [ "md5Hash" ] if "acl" in metadata : metadata [ "acl" ] = [ __postprocess_object_access_control ( bucket_id , metadata [ "name" ] , metadata [ "generation" ] , a ) for a in metadata [ "acl" ] ] return metadata
The protos for storage/v2 renamed some fields in ways that require some custom coding.
The protos for storage / v2 renamed some fields in ways that require some custom coding .
"""The protos for storage/v2 renamed some fields in ways that require some custom coding.""" # For some fields the storage/v2 name just needs to change slightly. # Checksums need special treatment # Finally the ACLs, if present, require additional fields
metadata