frostfs-testlib/src/frostfs_testlib/s3/aws_cli_client.py

import json
import logging
import os
import uuid
from datetime import datetime
from time import sleep
from typing import Literal, Optional, Union

from frostfs_testlib.reporter import get_reporter
from frostfs_testlib.resources.common import (
    ASSETS_DIR,
    MAX_REQUEST_ATTEMPTS,
    RETRY_MODE,
    S3_SYNC_WAIT_TIME,
)
from frostfs_testlib.s3.interfaces import S3ClientWrapper, VersioningStatus, _make_objs_dict

# TODO: Refactor this code to use shell instead of _cmd_run
from frostfs_testlib.utils.cli_utils import _cmd_run, _configure_aws_cli

reporter = get_reporter()
logger = logging.getLogger("NeoLogger")
LONG_TIMEOUT = 240


class AwsCliClient(S3ClientWrapper):
    __repr_name__: str = "AWS CLI"

    # Flags that we use for all S3 commands: disable SSL verification (as we use self-signed
    # certificate in devenv) and disable automatic pagination in CLI output
    common_flags = "--no-verify-ssl --no-paginate"
    s3gate_endpoint: str

    @reporter.step_deco("Configure S3 client (aws cli)")
    def __init__(self, access_key_id: str, secret_access_key: str, s3gate_endpoint: str) -> None:
        self.s3gate_endpoint = s3gate_endpoint
        try:
            _configure_aws_cli("aws configure", access_key_id, secret_access_key)
            _cmd_run(f"aws configure set max_attempts {MAX_REQUEST_ATTEMPTS}")
            _cmd_run(f"aws configure set retry_mode {RETRY_MODE}")
        except Exception as err:
            raise RuntimeError("Error while configuring AwsCliClient") from err

    @reporter.step_deco("Set endpoint S3 to {s3gate_endpoint}")
    def set_endpoint(self, s3gate_endpoint: str):
        self.s3gate_endpoint = s3gate_endpoint

    @reporter.step_deco("Create bucket S3")
    def create_bucket(
        self,
        bucket: Optional[str] = None,
        object_lock_enabled_for_bucket: Optional[bool] = None,
        acl: Optional[str] = None,
        grant_write: Optional[str] = None,
        grant_read: Optional[str] = None,
        grant_full_control: Optional[str] = None,
        location_constraint: Optional[str] = None,
    ) -> str:
        if bucket is None:
            bucket = str(uuid.uuid4())

        if object_lock_enabled_for_bucket is None:
            object_lock = ""
        elif object_lock_enabled_for_bucket:
            object_lock = " --object-lock-enabled-for-bucket"
        else:
            object_lock = " --no-object-lock-enabled-for-bucket"
        cmd = (
            f"aws {self.common_flags} s3api create-bucket --bucket {bucket} "
            f"{object_lock} --endpoint {self.s3gate_endpoint}"
        )
        if acl:
            cmd += f" --acl {acl}"
        if grant_full_control:
            cmd += f" --grant-full-control {grant_full_control}"
        if grant_write:
            cmd += f" --grant-write {grant_write}"
        if grant_read:
            cmd += f" --grant-read {grant_read}"
        if location_constraint:
            cmd += f" --create-bucket-configuration LocationConstraint={location_constraint}"
        _cmd_run(cmd)
        sleep(S3_SYNC_WAIT_TIME)

        return bucket

    @reporter.step_deco("List buckets S3")
    def list_buckets(self) -> list[str]:
        cmd = f"aws {self.common_flags} s3api list-buckets --endpoint {self.s3gate_endpoint}"
        output = _cmd_run(cmd)
        buckets_json = self._to_json(output)
        return [bucket["Name"] for bucket in buckets_json["Buckets"]]

    @reporter.step_deco("Delete bucket S3")
    def delete_bucket(self, bucket: str) -> None:
        cmd = f"aws {self.common_flags} s3api delete-bucket --bucket {bucket} --endpoint {self.s3gate_endpoint}"
        _cmd_run(cmd, LONG_TIMEOUT)
        sleep(S3_SYNC_WAIT_TIME)

    @reporter.step_deco("Head bucket S3")
    def head_bucket(self, bucket: str) -> None:
        cmd = f"aws {self.common_flags} s3api head-bucket --bucket {bucket} --endpoint {self.s3gate_endpoint}"
        _cmd_run(cmd)

    @reporter.step_deco("Put bucket versioning status")
    def put_bucket_versioning(self, bucket: str, status: VersioningStatus) -> None:
        cmd = (
            f"aws {self.common_flags} s3api put-bucket-versioning --bucket {bucket} "
            f"--versioning-configuration Status={status.value} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Get bucket versioning status")
    def get_bucket_versioning_status(self, bucket: str) -> Literal["Enabled", "Suspended"]:
        cmd = (
            f"aws {self.common_flags} s3api get-bucket-versioning --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("Status")

    @reporter.step_deco("Put bucket tagging")
    def put_bucket_tagging(self, bucket: str, tags: list) -> None:
        tags_json = {
            "TagSet": [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags]
        }
        cmd = (
            f"aws {self.common_flags} s3api put-bucket-tagging --bucket {bucket} "
            f"--tagging '{json.dumps(tags_json)}' --endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Get bucket tagging")
    def get_bucket_tagging(self, bucket: str) -> list:
        cmd = (
            f"aws {self.common_flags} s3api get-bucket-tagging --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("TagSet")

    @reporter.step_deco("Get bucket acl")
    def get_bucket_acl(self, bucket: str) -> list:
        cmd = (
            f"aws {self.common_flags} s3api get-bucket-acl --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("Grants")

    @reporter.step_deco("Get bucket location")
    def get_bucket_location(self, bucket: str) -> dict:
        cmd = (
            f"aws {self.common_flags} s3api get-bucket-location --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("LocationConstraint")

    @reporter.step_deco("List objects S3")
    def list_objects(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
        cmd = (
            f"aws {self.common_flags} s3api list-objects --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)

        obj_list = [obj["Key"] for obj in response.get("Contents", [])]
        logger.info(f"Found s3 objects: {obj_list}")

        return response if full_output else obj_list

    @reporter.step_deco("List objects S3 v2")
    def list_objects_v2(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
        cmd = (
            f"aws {self.common_flags} s3api list-objects-v2 --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)

        obj_list = [obj["Key"] for obj in response.get("Contents", [])]
        logger.info(f"Found s3 objects: {obj_list}")

        return response if full_output else obj_list

    @reporter.step_deco("List objects versions S3")
    def list_objects_versions(self, bucket: str, full_output: bool = False) -> dict:
        cmd = (
            f"aws {self.common_flags} s3api list-object-versions --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response if full_output else response.get("Versions", [])

    @reporter.step_deco("List objects delete markers S3")
    def list_delete_markers(self, bucket: str, full_output: bool = False) -> list:
        cmd = (
            f"aws {self.common_flags} s3api list-object-versions --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response if full_output else response.get("DeleteMarkers", [])

    @reporter.step_deco("Copy object S3")
    def copy_object(
        self,
        source_bucket: str,
        source_key: str,
        bucket: Optional[str] = None,
        key: Optional[str] = None,
        acl: Optional[str] = None,
        metadata_directive: Optional[Literal["COPY", "REPLACE"]] = None,
        metadata: Optional[dict] = None,
        tagging_directive: Optional[Literal["COPY", "REPLACE"]] = None,
        tagging: Optional[str] = None,
    ) -> str:
        if bucket is None:
            bucket = source_bucket
        if key is None:
            key = os.path.join(os.getcwd(), str(uuid.uuid4()))
        copy_source = f"{source_bucket}/{source_key}"

        cmd = (
            f"aws {self.common_flags} s3api copy-object --copy-source {copy_source} "
            f"--bucket {bucket} --key {key} --endpoint {self.s3gate_endpoint}"
        )
        if acl:
            cmd += f" --acl {acl}"
        if metadata_directive:
            cmd += f" --metadata-directive {metadata_directive}"
        if metadata:
            cmd += " --metadata "
            for meta_key, value in metadata.items():
                cmd += f" {meta_key}={value}"
        if tagging_directive:
            cmd += f" --tagging-directive {tagging_directive}"
        if tagging:
            cmd += f" --tagging {tagging}"
        _cmd_run(cmd, LONG_TIMEOUT)
        return key

    @reporter.step_deco("Put object S3")
    def put_object(
        self,
        bucket: str,
        filepath: str,
        key: Optional[str] = None,
        metadata: Optional[dict] = None,
        tagging: Optional[str] = None,
        acl: Optional[str] = None,
        object_lock_mode: Optional[str] = None,
        object_lock_retain_until_date: Optional[datetime] = None,
        object_lock_legal_hold_status: Optional[str] = None,
        grant_full_control: Optional[str] = None,
        grant_read: Optional[str] = None,
    ) -> str:
        if key is None:
            key = os.path.basename(filepath)

        cmd = (
            f"aws {self.common_flags} s3api put-object --bucket {bucket} --key {key} "
            f"--body {filepath} --endpoint {self.s3gate_endpoint}"
        )
        if metadata:
            cmd += " --metadata"
            for key, value in metadata.items():
                cmd += f" {key}={value}"
        if tagging:
            cmd += f" --tagging '{tagging}'"
        if acl:
            cmd += f" --acl {acl}"
        if object_lock_mode:
            cmd += f" --object-lock-mode {object_lock_mode}"
        if object_lock_retain_until_date:
            cmd += f' --object-lock-retain-until-date "{object_lock_retain_until_date}"'
        if object_lock_legal_hold_status:
            cmd += f" --object-lock-legal-hold-status {object_lock_legal_hold_status}"
        if grant_full_control:
            cmd += f" --grant-full-control '{grant_full_control}'"
        if grant_read:
            cmd += f" --grant-read {grant_read}"
        output = _cmd_run(cmd, LONG_TIMEOUT)
        response = self._to_json(output)
        return response.get("VersionId")

    @reporter.step_deco("Head object S3")
    def head_object(self, bucket: str, key: str, version_id: Optional[str] = None) -> dict:
        version = f" --version-id {version_id}" if version_id else ""
        cmd = (
            f"aws {self.common_flags} s3api head-object --bucket {bucket} --key {key} "
            f"{version} --endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response

    @reporter.step_deco("Get object S3")
    def get_object(
        self,
        bucket: str,
        key: str,
        version_id: Optional[str] = None,
        object_range: Optional[tuple[int, int]] = None,
        full_output: bool = False,
    ) -> Union[dict, str]:
        file_path = os.path.join(os.getcwd(), ASSETS_DIR, str(uuid.uuid4()))
        version = f" --version-id {version_id}" if version_id else ""
        cmd = (
            f"aws {self.common_flags} s3api get-object --bucket {bucket} --key {key} "
            f"{version} {file_path} --endpoint {self.s3gate_endpoint}"
        )
        if object_range:
            cmd += f" --range bytes={object_range[0]}-{object_range[1]}"
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response if full_output else file_path

    @reporter.step_deco("Get object ACL")
    def get_object_acl(self, bucket: str, key: str, version_id: Optional[str] = None) -> list:
        version = f" --version-id {version_id}" if version_id else ""
        cmd = (
            f"aws {self.common_flags} s3api get-object-acl --bucket {bucket} --key {key} "
            f"{version} --endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("Grants")

    @reporter.step_deco("Put object ACL")
    def put_object_acl(
        self,
        bucket: str,
        key: str,
        acl: Optional[str] = None,
        grant_write: Optional[str] = None,
        grant_read: Optional[str] = None,
    ) -> list:
        cmd = (
            f"aws {self.common_flags} s3api put-object-acl --bucket {bucket} --key {key} "
            f" --endpoint {self.s3gate_endpoint}"
        )
        if acl:
            cmd += f" --acl {acl}"
        if grant_write:
            cmd += f" --grant-write {grant_write}"
        if grant_read:
            cmd += f" --grant-read {grant_read}"
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("Grants")

    @reporter.step_deco("Put bucket ACL")
    def put_bucket_acl(
        self,
        bucket: str,
        acl: Optional[str] = None,
        grant_write: Optional[str] = None,
        grant_read: Optional[str] = None,
    ) -> None:
        cmd = (
            f"aws {self.common_flags} s3api put-bucket-acl --bucket {bucket} "
            f" --endpoint {self.s3gate_endpoint}"
        )
        if acl:
            cmd += f" --acl {acl}"
        if grant_write:
            cmd += f" --grant-write {grant_write}"
        if grant_read:
            cmd += f" --grant-read {grant_read}"
        _cmd_run(cmd)

    @reporter.step_deco("Delete objects S3")
    def delete_objects(self, bucket: str, keys: list[str]) -> dict:
        file_path = os.path.join(os.getcwd(), ASSETS_DIR, "delete.json")
        delete_structure = json.dumps(_make_objs_dict(keys))
        with open(file_path, "w") as out_file:
            out_file.write(delete_structure)
        logger.info(f"Input file for delete-objects: {delete_structure}")

        cmd = (
            f"aws {self.common_flags} s3api delete-objects --bucket {bucket} "
            f"--delete file://{file_path} --endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd, LONG_TIMEOUT)
        response = self._to_json(output)
        sleep(S3_SYNC_WAIT_TIME)
        return response

    @reporter.step_deco("Delete object S3")
    def delete_object(self, bucket: str, key: str, version_id: Optional[str] = None) -> dict:
        version = f" --version-id {version_id}" if version_id else ""
        cmd = (
            f"aws {self.common_flags} s3api delete-object --bucket {bucket} "
            f"--key {key} {version} --endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd, LONG_TIMEOUT)
        sleep(S3_SYNC_WAIT_TIME)
        return self._to_json(output)

    @reporter.step_deco("Delete object versions S3")
    def delete_object_versions(self, bucket: str, object_versions: list) -> dict:
        # Build deletion list in S3 format
        delete_list = {
            "Objects": [
                {
                    "Key": object_version["Key"],
                    "VersionId": object_version["VersionId"],
                }
                for object_version in object_versions
            ]
        }

        file_path = os.path.join(os.getcwd(), ASSETS_DIR, "delete.json")
        delete_structure = json.dumps(delete_list)
        with open(file_path, "w") as out_file:
            out_file.write(delete_structure)
        logger.info(f"Input file for delete-objects: {delete_structure}")

        cmd = (
            f"aws {self.common_flags} s3api delete-objects --bucket {bucket} "
            f"--delete file://{file_path} --endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd, LONG_TIMEOUT)
        sleep(S3_SYNC_WAIT_TIME)
        return self._to_json(output)

    @reporter.step_deco("Delete object versions S3 without delete markers")
    def delete_object_versions_without_dm(self, bucket: str, object_versions: list) -> None:
        # Delete objects without creating delete markers
        for object_version in object_versions:
            self.delete_object(
                bucket=bucket, key=object_version["Key"], version_id=object_version["VersionId"]
            )

    @reporter.step_deco("Get object attributes")
    def get_object_attributes(
        self,
        bucket: str,
        key: str,
        attributes: list[str],
        version_id: str = "",
        max_parts: int = 0,
        part_number: int = 0,
        full_output: bool = True,
    ) -> dict:

        attrs = ",".join(attributes)
        version = f" --version-id {version_id}" if version_id else ""
        parts = f"--max-parts {max_parts}" if max_parts else ""
        part_number_str = f"--part-number-marker {part_number}" if part_number else ""
        cmd = (
            f"aws {self.common_flags} s3api get-object-attributes --bucket {bucket} "
            f"--key {key} {version} {parts} {part_number_str} --object-attributes {attrs} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)

        for attr in attributes:
            assert attr in response, f"Expected attribute {attr} in {response}"

        if full_output:
            return response
        else:
            return response.get(attributes[0])

    @reporter.step_deco("Get bucket policy")
    def get_bucket_policy(self, bucket: str) -> dict:
        cmd = (
            f"aws {self.common_flags} s3api get-bucket-policy --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("Policy")

    @reporter.step_deco("Put bucket policy")
    def put_bucket_policy(self, bucket: str, policy: dict) -> None:
        # Leaving it as is was in test repo. Double dumps to escape resulting string
        # Example:
        #  policy = {"a": 1}
        # json.dumps(policy) => {"a": 1}
        # json.dumps(json.dumps(policy)) => "{\"a\": 1}"
        # TODO: update this
        dumped_policy = json.dumps(json.dumps(policy))
        cmd = (
            f"aws {self.common_flags} s3api put-bucket-policy --bucket {bucket} "
            f"--policy {dumped_policy} --endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Get bucket cors")
    def get_bucket_cors(self, bucket: str) -> dict:
        cmd = (
            f"aws {self.common_flags} s3api get-bucket-cors --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("CORSRules")

    @reporter.step_deco("Put bucket cors")
    def put_bucket_cors(self, bucket: str, cors_configuration: dict) -> None:
        cmd = (
            f"aws {self.common_flags} s3api put-bucket-cors --bucket {bucket} "
            f"--cors-configuration '{json.dumps(cors_configuration)}' --endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Delete bucket cors")
    def delete_bucket_cors(self, bucket: str) -> None:
        cmd = (
            f"aws {self.common_flags} s3api delete-bucket-cors --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Delete bucket tagging")
    def delete_bucket_tagging(self, bucket: str) -> None:
        cmd = (
            f"aws {self.common_flags} s3api delete-bucket-tagging --bucket {bucket} "
            f"--endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Put object retention")
    def put_object_retention(
        self,
        bucket: str,
        key: str,
        retention: dict,
        version_id: Optional[str] = None,
        bypass_governance_retention: Optional[bool] = None,
    ) -> None:
        version = f" --version-id {version_id}" if version_id else ""
        cmd = (
            f"aws {self.common_flags} s3api put-object-retention --bucket {bucket} --key {key} "
            f"{version} --retention '{json.dumps(retention, indent=4, sort_keys=True, default=str)}' --endpoint {self.s3gate_endpoint}"
        )
        if bypass_governance_retention is not None:
            cmd += " --bypass-governance-retention"
        _cmd_run(cmd)

    @reporter.step_deco("Put object legal hold")
    def put_object_legal_hold(
        self,
        bucket: str,
        key: str,
        legal_hold_status: Literal["ON", "OFF"],
        version_id: Optional[str] = None,
    ) -> None:
        version = f" --version-id {version_id}" if version_id else ""
        legal_hold = json.dumps({"Status": legal_hold_status})
        cmd = (
            f"aws {self.common_flags} s3api  put-object-legal-hold --bucket {bucket} --key {key} "
            f"{version} --legal-hold '{legal_hold}' --endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Put object tagging")
    def put_object_tagging(self, bucket: str, key: str, tags: list) -> None:
        tags = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags]
        tagging = {"TagSet": tags}
        cmd = (
            f"aws {self.common_flags} s3api put-object-tagging --bucket {bucket} --key {key} "
            f"--tagging '{json.dumps(tagging)}' --endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Get object tagging")
    def get_object_tagging(self, bucket: str, key: str, version_id: Optional[str] = None) -> list:
        version = f" --version-id {version_id}" if version_id else ""
        cmd = (
            f"aws {self.common_flags} s3api get-object-tagging --bucket {bucket} --key {key} "
            f"{version}  --endpoint {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("TagSet")

    @reporter.step_deco("Delete object tagging")
    def delete_object_tagging(self, bucket: str, key: str) -> None:
        cmd = (
            f"aws {self.common_flags} s3api delete-object-tagging --bucket {bucket} "
            f"--key {key} --endpoint {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Sync directory S3")
    def sync(
        self,
        bucket: str,
        dir_path: str,
        acl: Optional[str] = None,
        metadata: Optional[dict] = None,
    ) -> dict:
        cmd = (
            f"aws {self.common_flags} s3 sync {dir_path}  s3://{bucket} "
            f"--endpoint-url {self.s3gate_endpoint}"
        )
        if metadata:
            cmd += " --metadata"
            for key, value in metadata.items():
                cmd += f" {key}={value}"
        if acl:
            cmd += f" --acl {acl}"
        output = _cmd_run(cmd, LONG_TIMEOUT)
        return self._to_json(output)

    @reporter.step_deco("CP directory S3")
    def cp(
        self,
        bucket: str,
        dir_path: str,
        acl: Optional[str] = None,
        metadata: Optional[dict] = None,
    ) -> dict:
        cmd = (
            f"aws {self.common_flags} s3 cp {dir_path}  s3://{bucket} "
            f"--endpoint-url {self.s3gate_endpoint}  --recursive"
        )
        if metadata:
            cmd += " --metadata"
            for key, value in metadata.items():
                cmd += f" {key}={value}"
        if acl:
            cmd += f" --acl {acl}"
        output = _cmd_run(cmd, LONG_TIMEOUT)
        return self._to_json(output)

    @reporter.step_deco("Create multipart upload S3")
    def create_multipart_upload(self, bucket: str, key: str) -> str:
        cmd = (
            f"aws {self.common_flags} s3api create-multipart-upload --bucket {bucket} "
            f"--key {key} --endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)

        assert response.get("UploadId"), f"Expected UploadId in response:\n{response}"

        return response["UploadId"]

    @reporter.step_deco("List multipart uploads S3")
    def list_multipart_uploads(self, bucket: str) -> Optional[list[dict]]:
        cmd = (
            f"aws {self.common_flags} s3api list-multipart-uploads --bucket {bucket} "
            f"--endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("Uploads")

    @reporter.step_deco("Abort multipart upload S3")
    def abort_multipart_upload(self, bucket: str, key: str, upload_id: str) -> None:
        cmd = (
            f"aws {self.common_flags} s3api abort-multipart-upload  --bucket {bucket} "
            f"--key {key} --upload-id {upload_id} --endpoint-url {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Upload part S3")
    def upload_part(
        self, bucket: str, key: str, upload_id: str, part_num: int, filepath: str
    ) -> str:
        cmd = (
            f"aws {self.common_flags} s3api upload-part --bucket {bucket} --key {key} "
            f"--upload-id {upload_id} --part-number {part_num} --body {filepath} "
            f"--endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd, LONG_TIMEOUT)
        response = self._to_json(output)
        assert response.get("ETag"), f"Expected ETag in response:\n{response}"
        return response["ETag"]

    @reporter.step_deco("Upload copy part S3")
    def upload_part_copy(
        self, bucket: str, key: str, upload_id: str, part_num: int, copy_source: str
    ) -> str:
        cmd = (
            f"aws {self.common_flags} s3api upload-part-copy --bucket {bucket} --key {key} "
            f"--upload-id {upload_id} --part-number {part_num} --copy-source {copy_source} "
            f"--endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd, LONG_TIMEOUT)
        response = self._to_json(output)
        assert response.get("CopyPartResult", []).get(
            "ETag"
        ), f"Expected ETag in response:\n{response}"

        return response["CopyPartResult"]["ETag"]

    @reporter.step_deco("List parts S3")
    def list_parts(self, bucket: str, key: str, upload_id: str) -> list[dict]:
        cmd = (
            f"aws {self.common_flags} s3api list-parts --bucket {bucket} --key {key} "
            f"--upload-id {upload_id} --endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)

        assert response.get("Parts"), f"Expected Parts in response:\n{response}"

        return response["Parts"]

    @reporter.step_deco("Complete multipart upload S3")
    def complete_multipart_upload(self, bucket: str, key: str, upload_id: str, parts: list) -> None:
        file_path = os.path.join(os.getcwd(), ASSETS_DIR, "parts.json")
        parts_dict = {"Parts": [{"ETag": etag, "PartNumber": part_num} for part_num, etag in parts]}

        with open(file_path, "w") as out_file:
            out_file.write(json.dumps(parts_dict))

        logger.info(f"Input file for complete-multipart-upload: {json.dumps(parts_dict)}")

        cmd = (
            f"aws {self.common_flags} s3api complete-multipart-upload --bucket {bucket} "
            f"--key {key}  --upload-id {upload_id} --multipart-upload file://{file_path} "
            f"--endpoint-url {self.s3gate_endpoint}"
        )
        _cmd_run(cmd)

    @reporter.step_deco("Put object lock configuration")
    def put_object_lock_configuration(self, bucket: str, configuration: dict) -> dict:
        cmd = (
            f"aws {self.common_flags} s3api put-object-lock-configuration --bucket {bucket} "
            f"--object-lock-configuration '{json.dumps(configuration)}' --endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        return self._to_json(output)

    @reporter.step_deco("Get object lock configuration")
    def get_object_lock_configuration(self, bucket: str):
        cmd = (
            f"aws {self.common_flags} s3api get-object-lock-configuration --bucket {bucket} "
            f"--endpoint-url {self.s3gate_endpoint}"
        )
        output = _cmd_run(cmd)
        response = self._to_json(output)
        return response.get("ObjectLockConfiguration")

    @staticmethod
    def _to_json(output: str) -> dict:
        json_output = {}
        if "{" not in output and "}" not in output:
            logger.warning(f"Could not parse json from output {output}")
            return json_output

        json_output = json.loads(output[output.index("{") :])

        return json_output