import logging import uuid from enum import Enum from time import sleep from typing import Optional import allure from botocore.exceptions import ClientError from cli_helpers import log_command_execution logger = logging.getLogger("NeoLogger") # Artificial delay that we add after object deletion and container creation # Delay is added because sometimes immediately after deletion object still appears # to be existing (probably because tombstone object takes some time to replicate) # TODO: remove after https://github.com/nspcc-dev/neofs-s3-gw/issues/610 is fixed S3_SYNC_WAIT_TIME = 5 class VersioningStatus(Enum): ENABLED = "Enabled" SUSPENDED = "Suspended" @allure.step("Create bucket S3") def create_bucket_s3( s3_client, object_lock_enabled_for_bucket: Optional[bool] = None, acl: Optional[str] = None ) -> str: bucket_name = str(uuid.uuid4()) try: params = {"Bucket": bucket_name} if object_lock_enabled_for_bucket is not None: params.update({"ObjectLockEnabledForBucket": object_lock_enabled_for_bucket}) if acl is not None: params.update({"ACL": acl}) s3_bucket = s3_client.create_bucket(**params) log_command_execution(f"Created S3 bucket {bucket_name}", s3_bucket) sleep(S3_SYNC_WAIT_TIME) return bucket_name except ClientError as err: raise Exception( f'Error Message: {err.response["Error"]["Message"]}\n' f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}' ) from err @allure.step("List buckets S3") def list_buckets_s3(s3_client): found_buckets = [] try: response = s3_client.list_buckets() log_command_execution("S3 List buckets result", response) for bucket in response["Buckets"]: found_buckets.append(bucket["Name"]) return found_buckets except ClientError as err: raise Exception( f'Error Message: {err.response["Error"]["Message"]}\n' f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}' ) from err @allure.step("Delete bucket S3") def delete_bucket_s3(s3_client, bucket: str): try: response = s3_client.delete_bucket(Bucket=bucket) log_command_execution("S3 Delete bucket result", response) sleep(S3_SYNC_WAIT_TIME) return response except ClientError as err: log_command_execution("S3 Delete bucket error", str(err)) raise Exception( f'Error Message: {err.response["Error"]["Message"]}\n' f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}' ) from err @allure.step("Head bucket S3") def head_bucket(s3_client, bucket: str): try: response = s3_client.head_bucket(Bucket=bucket) log_command_execution("S3 Head bucket result", response) return response except ClientError as err: log_command_execution("S3 Head bucket error", str(err)) raise Exception( f'Error Message: {err.response["Error"]["Message"]}\n' f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}' ) from err @allure.step("Set bucket versioning status") def set_bucket_versioning(s3_client, bucket_name: str, status: VersioningStatus) -> None: try: response = s3_client.put_bucket_versioning( Bucket=bucket_name, VersioningConfiguration={"Status": status.value} ) log_command_execution("S3 Set bucket versioning to", response) except ClientError as err: raise Exception(f"Got error during set bucket versioning: {err}") from err @allure.step("Get bucket versioning status") def get_bucket_versioning_status(s3_client, bucket_name: str) -> str: try: response = s3_client.get_bucket_versioning(Bucket=bucket_name) status = response.get("Status") log_command_execution("S3 Got bucket versioning status", response) return status except ClientError as err: raise Exception(f"Got error during get bucket versioning status: {err}") from err @allure.step("Put bucket tagging") def put_bucket_tagging(s3_client, bucket_name: str, tags: list): try: tags = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags] tagging = {"TagSet": tags} response = s3_client.put_bucket_tagging(Bucket=bucket_name, Tagging=tagging) log_command_execution("S3 Put bucket tagging", response) except ClientError as err: raise Exception(f"Got error during put bucket tagging: {err}") from err @allure.step("Get bucket acl") def get_bucket_acl(s3_client, bucket_name: str) -> list: try: response = s3_client.get_bucket_acl(Bucket=bucket_name) log_command_execution("S3 Get bucket acl", response) return response.get("Grants") except ClientError as err: raise Exception(f"Got error during get bucket tagging: {err}") from err @allure.step("Get bucket tagging") def get_bucket_tagging(s3_client, bucket_name: str) -> list: try: response = s3_client.get_bucket_tagging(Bucket=bucket_name) log_command_execution("S3 Get bucket tagging", response) return response.get("TagSet") except ClientError as err: raise Exception(f"Got error during get bucket tagging: {err}") from err @allure.step("Delete bucket tagging") def delete_bucket_tagging(s3_client, bucket_name: str) -> None: try: response = s3_client.delete_bucket_tagging(Bucket=bucket_name) log_command_execution("S3 Delete bucket tagging", response) except ClientError as err: raise Exception(f"Got error during delete bucket tagging: {err}") from err @allure.step("Put bucket ACL") def put_bucket_acl_s3( s3_client, bucket: str, acl: Optional[str] = None, grant_write: Optional[str] = None, grant_read: Optional[str] = None, ) -> list: params = {"Bucket": bucket} if acl: params.update({"ACL": acl}) elif grant_write or grant_read: if grant_write: params.update({"GrantWrite": grant_write}) elif grant_read: params.update({"GrantRead": grant_read}) try: response = s3_client.put_bucket_acl(**params) log_command_execution("S3 ACL bucket result", response) return response.get("Grants") except ClientError as err: raise Exception( f'Error Message: {err.response["Error"]["Message"]}\n' f'Http status code: {err.response["ResponseMetadata"]["HTTPStatusCode"]}' ) from err