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 from frostfs_testlib.shell import CommandOptions from frostfs_testlib.shell.local_shell import LocalShell # TODO: Refactor this code to use shell instead of _cmd_run from frostfs_testlib.utils.cli_utils import _configure_aws_cli reporter = get_reporter() logger = logging.getLogger("NeoLogger") command_options = CommandOptions(timeout=480) 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, profile: str='default') -> None: self.s3gate_endpoint = s3gate_endpoint self.profile = profile self.local_shell = LocalShell() try: _configure_aws_cli(f"aws configure --profile {profile}", access_key_id, secret_access_key) self.local_shell.exec(f"aws configure set max_attempts {MAX_REQUEST_ATTEMPTS} --profile {profile}") self.local_shell.exec( f"aws configure set retry_mode {RETRY_MODE} --profile {profile}", ) 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} --profile {self.profile}" ) 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}" self.local_shell.exec(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} --profile {self.profile}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" self.local_shell.exec(cmd, command_options) 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} --profile {self.profile}" self.local_shell.exec(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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) 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}" self.local_shell.exec(cmd, command_options) 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} --profile {self.profile}" ) 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 = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) if object_range: cmd += f" --range bytes={object_range[0]}-{object_range[1]}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) 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 = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" if acl: cmd += f" --acl {acl}" if grant_write: cmd += f" --grant-write {grant_write}" if grant_read: cmd += f" --grant-read {grant_read}" self.local_shell.exec(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} --profile {self.profile}" ) output = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) if bypass_governance_retention is not None: cmd += " --bypass-governance-retention" self.local_shell.exec(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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" if metadata: cmd += " --metadata" for key, value in metadata.items(): cmd += f" {key}={value}" if acl: cmd += f" --acl {acl}" output = self.local_shell.exec(cmd, command_options).stdout 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 --profile {self.profile}" ) if metadata: cmd += " --metadata" for key, value in metadata.items(): cmd += f" {key}={value}" if acl: cmd += f" --acl {acl}" output = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) output = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd, command_options).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) self.local_shell.exec(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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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} --profile {self.profile}" ) output = self.local_shell.exec(cmd).stdout 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