Added delete bucket policy method to s3 client

This commit is contained in:
Yaroslava Lukoyanova 2024-05-16 10:13:11 +03:00
parent cc13a43bec
commit f236c1b083
3 changed files with 21 additions and 0 deletions

View file

@ -489,6 +489,16 @@ class AwsCliClient(S3ClientWrapper):
response = self._to_json(output) response = self._to_json(output)
return response.get("Policy") return response.get("Policy")
@reporter.step("Delete bucket policy")
def delete_bucket_policy(self, bucket: str) -> dict:
cmd = (
f"aws {self.common_flags} s3api delete-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
@reporter.step("Put bucket policy") @reporter.step("Put bucket policy")
def put_bucket_policy(self, bucket: str, policy: dict) -> None: def put_bucket_policy(self, bucket: str, policy: dict) -> None:
# Leaving it as is was in test repo. Double dumps to escape resulting string # Leaving it as is was in test repo. Double dumps to escape resulting string

View file

@ -246,6 +246,13 @@ class Boto3ClientWrapper(S3ClientWrapper):
log_command_execution("S3 get_bucket_policy result", response) log_command_execution("S3 get_bucket_policy result", response)
return response.get("Policy") return response.get("Policy")
@reporter.step("Delete bucket policy")
@report_error
def delete_bucket_policy(self, bucket: str) -> str:
response = self.boto3_client.delete_bucket_policy(Bucket=bucket)
log_command_execution("S3 delete_bucket_policy result", response)
return response
@reporter.step("Put bucket policy") @reporter.step("Put bucket policy")
@report_error @report_error
def put_bucket_policy(self, bucket: str, policy: dict) -> None: def put_bucket_policy(self, bucket: str, policy: dict) -> None:

View file

@ -152,6 +152,10 @@ class S3ClientWrapper(HumanReadableABC):
def get_bucket_policy(self, bucket: str) -> str: def get_bucket_policy(self, bucket: str) -> str:
"""Returns the policy of a specified bucket.""" """Returns the policy of a specified bucket."""
@abstractmethod
def delete_bucket_policy(self, bucket: str) -> str:
"""Deletes the policy of a specified bucket."""
@abstractmethod @abstractmethod
def put_bucket_policy(self, bucket: str, policy: dict) -> None: def put_bucket_policy(self, bucket: str, policy: dict) -> None:
"""Applies S3 bucket policy to an S3 bucket.""" """Applies S3 bucket policy to an S3 bucket."""