TrueCloudLab/frostfs-testlib
24
0
Fork 18
Code Issues Pull requests 2 Releases 1 Activity Actions

[#259] Improve logging of boto3 client requests
Some checks reported warnings
DCO action / DCO (pull_request) Has been cancelled
Details

Browse source 
Signed-off-by: Kirill Sosnovskikh <k.sosnovskikh@yadro.com>
This commit is contained in:
k.sosnovskikh 2024-07-10 17:17:27 +03:00
parent 429698944e
commit aef16c5e44
2 changed files with 123 additions and 106 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

219
src/frostfs_testlib/s3/boto3_client.py
View file

@ -18,7 +18,7 @@ from frostfs_testlib.s3.interfaces import S3ClientWrapper, VersioningStatus, _ma
from frostfs_testlib.utils import string_utils from frostfs_testlib.utils import string_utils
# TODO: Refactor this code to use shell instead of _cmd_run # TODO: Refactor this code to use shell instead of _cmd_run
from frostfs_testlib.utils.cli_utils import _configure_aws_cli, log_command_execution from frostfs_testlib.utils.cli_utils import log_command_execution
from frostfs_testlib.utils.file_utils import TestFile from frostfs_testlib.utils.file_utils import TestFile
logger = logging.getLogger("NeoLogger") logger = logging.getLogger("NeoLogger")
@ -34,7 +34,15 @@ def report_error(func):
try: try:
return func(*a, **kw) return func(*a, **kw)
except ClientError as err: except ClientError as err:
log_command_execution("Result", str(err)) url = None
params = {"args": a, "kwargs": kw}
if isinstance(a[0], Boto3ClientWrapper):
client: Boto3ClientWrapper = a[0]
url = client.s3gate_endpoint
params = {"args": a[1:], "kwargs": kw}
log_command_execution(url, f"Failed {err.operation_name}", err.response, params)
raise raise
return deco return deco
@ -101,6 +109,11 @@ class Boto3ClientWrapper(S3ClientWrapper):
result = result.replace(find, replace) result = result.replace(find, replace)
return result return result
def _convert_to_s3_params(self, scope: dict, exclude: list[str] = None) -> dict:
if not exclude:
exclude = ["self"]
return {self._to_s3_param(param): value for param, value in scope if param not in exclude and value is not None}
# BUCKET METHODS # # BUCKET METHODS #
@reporter.step("Create bucket S3") @reporter.step("Create bucket S3")
@report_error @report_error
@ -133,7 +146,7 @@ class Boto3ClientWrapper(S3ClientWrapper):
params.update({"CreateBucketConfiguration": {"LocationConstraint": location_constraint}}) params.update({"CreateBucketConfiguration": {"LocationConstraint": location_constraint}})
s3_bucket = self.boto3_client.create_bucket(**params) s3_bucket = self.boto3_client.create_bucket(**params)
log_command_execution(f"Created S3 bucket {bucket}", s3_bucket) log_command_execution(self.s3gate_endpoint, f"Created S3 bucket {bucket}", s3_bucket, params)
return bucket return bucket
@reporter.step("List buckets S3") @reporter.step("List buckets S3")
@ -142,7 +155,7 @@ class Boto3ClientWrapper(S3ClientWrapper):
found_buckets = [] found_buckets = []
response = self.boto3_client.list_buckets() response = self.boto3_client.list_buckets()
log_command_execution("S3 List buckets result", response) log_command_execution(self.s3gate_endpoint, "S3 List buckets result", response)
for bucket in response["Buckets"]: for bucket in response["Buckets"]:
found_buckets.append(bucket["Name"]) found_buckets.append(bucket["Name"])
@ -153,26 +166,27 @@ class Boto3ClientWrapper(S3ClientWrapper):
@report_error @report_error
def delete_bucket(self, bucket: str) -> None: def delete_bucket(self, bucket: str) -> None:
response = self.boto3_client.delete_bucket(Bucket=bucket) response = self.boto3_client.delete_bucket(Bucket=bucket)
log_command_execution("S3 Delete bucket result", response) log_command_execution(self.s3gate_endpoint, "S3 Delete bucket result", response, {"Bucket": bucket})
@reporter.step("Head bucket S3") @reporter.step("Head bucket S3")
@report_error @report_error
def head_bucket(self, bucket: str) -> None: def head_bucket(self, bucket: str) -> None:
response = self.boto3_client.head_bucket(Bucket=bucket) response = self.boto3_client.head_bucket(Bucket=bucket)
log_command_execution("S3 Head bucket result", response) log_command_execution(self.s3gate_endpoint, "S3 Head bucket result", response, {"Bucket": bucket})
@reporter.step("Put bucket versioning status") @reporter.step("Put bucket versioning status")
@report_error @report_error
def put_bucket_versioning(self, bucket: str, status: VersioningStatus) -> None: def put_bucket_versioning(self, bucket: str, status: VersioningStatus) -> None:
response = self.boto3_client.put_bucket_versioning(Bucket=bucket, VersioningConfiguration={"Status": status.value}) params = {"Bucket": bucket, "VersioningConfiguration": {"Status": status.value}}
log_command_execution("S3 Set bucket versioning to", response) response = self.boto3_client.put_bucket_versioning(**params)
log_command_execution(self.s3gate_endpoint, "S3 Set bucket versioning to", response, params)
@reporter.step("Get bucket versioning status") @reporter.step("Get bucket versioning status")
@report_error @report_error
def get_bucket_versioning_status(self, bucket: str) -> Literal["Enabled", "Suspended"]: def get_bucket_versioning_status(self, bucket: str) -> Literal["Enabled", "Suspended"]:
response = self.boto3_client.get_bucket_versioning(Bucket=bucket) response = self.boto3_client.get_bucket_versioning(Bucket=bucket)
status = response.get("Status") status = response.get("Status")
log_command_execution("S3 Got bucket versioning status", response) log_command_execution(self.s3gate_endpoint, "S3 Got bucket versioning status", response, {"Bucket": bucket})
return status return status
@reporter.step("Put bucket tagging") @reporter.step("Put bucket tagging")
@ -180,28 +194,29 @@ class Boto3ClientWrapper(S3ClientWrapper):
def put_bucket_tagging(self, bucket: str, tags: list) -> None: def put_bucket_tagging(self, bucket: str, tags: list) -> None:
tags = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags] tags = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags]
tagging = {"TagSet": tags} tagging = {"TagSet": tags}
response = self.boto3_client.put_bucket_tagging(Bucket=bucket, Tagging=tagging) params = self._convert_to_s3_params(locals().items(), exclude=["self", "tags"])
log_command_execution("S3 Put bucket tagging", response) response = self.boto3_client.put_bucket_tagging(**params)
log_command_execution(self.s3gate_endpoint, "S3 Put bucket tagging", response, params)
@reporter.step("Get bucket tagging") @reporter.step("Get bucket tagging")
@report_error @report_error
def get_bucket_tagging(self, bucket: str) -> list: def get_bucket_tagging(self, bucket: str) -> list:
response = self.boto3_client.get_bucket_tagging(Bucket=bucket) response = self.boto3_client.get_bucket_tagging(Bucket=bucket)
log_command_execution("S3 Get bucket tagging", response) log_command_execution(self.s3gate_endpoint, "S3 Get bucket tagging", response, {"Bucket": bucket})
return response.get("TagSet") return response.get("TagSet")
@reporter.step("Get bucket acl") @reporter.step("Get bucket acl")
@report_error @report_error
def get_bucket_acl(self, bucket: str) -> list: def get_bucket_acl(self, bucket: str) -> list:
response = self.boto3_client.get_bucket_acl(Bucket=bucket) response = self.boto3_client.get_bucket_acl(Bucket=bucket)
log_command_execution("S3 Get bucket acl", response) log_command_execution(self.s3gate_endpoint, "S3 Get bucket acl", response, {"Bucket": bucket})
return response.get("Grants") return response.get("Grants")
@reporter.step("Delete bucket tagging") @reporter.step("Delete bucket tagging")
@report_error @report_error
def delete_bucket_tagging(self, bucket: str) -> None: def delete_bucket_tagging(self, bucket: str) -> None:
response = self.boto3_client.delete_bucket_tagging(Bucket=bucket) response = self.boto3_client.delete_bucket_tagging(Bucket=bucket)
log_command_execution("S3 Delete bucket tagging", response) log_command_execution(self.s3gate_endpoint, "S3 Delete bucket tagging", response, {"Bucket": bucket})
@reporter.step("Put bucket ACL") @reporter.step("Put bucket ACL")
@report_error @report_error
@ -212,71 +227,74 @@ class Boto3ClientWrapper(S3ClientWrapper):
grant_write: Optional[str] = None, grant_write: Optional[str] = None,
grant_read: Optional[str] = None, grant_read: Optional[str] = None,
) -> None: ) -> None:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.put_bucket_acl(**params) response = self.boto3_client.put_bucket_acl(**params)
log_command_execution("S3 ACL bucket result", response) log_command_execution(self.s3gate_endpoint, "S3 ACL bucket result", response, params)
@reporter.step("Put object lock configuration") @reporter.step("Put object lock configuration")
@report_error @report_error
def put_object_lock_configuration(self, bucket: str, configuration: dict) -> dict: def put_object_lock_configuration(self, bucket: str, configuration: dict) -> dict:
response = self.boto3_client.put_object_lock_configuration(Bucket=bucket, ObjectLockConfiguration=configuration) params = {"Bucket": bucket, "ObjectLockConfiguration": configuration}
log_command_execution("S3 put_object_lock_configuration result", response) response = self.boto3_client.put_object_lock_configuration(**params)
log_command_execution(self.s3gate_endpoint, "S3 put_object_lock_configuration result", response, params)
return response return response
@reporter.step("Get object lock configuration") @reporter.step("Get object lock configuration")
@report_error @report_error
def get_object_lock_configuration(self, bucket: str) -> dict: def get_object_lock_configuration(self, bucket: str) -> dict:
response = self.boto3_client.get_object_lock_configuration(Bucket=bucket) response = self.boto3_client.get_object_lock_configuration(Bucket=bucket)
log_command_execution("S3 get_object_lock_configuration result", response) log_command_execution(self.s3gate_endpoint, "S3 get_object_lock_configuration result", response, {"Bucket": bucket})
return response.get("ObjectLockConfiguration") return response.get("ObjectLockConfiguration")
@reporter.step("Get bucket policy") @reporter.step("Get bucket policy")
@report_error @report_error
def get_bucket_policy(self, bucket: str) -> str: def get_bucket_policy(self, bucket: str) -> str:
response = self.boto3_client.get_bucket_policy(Bucket=bucket) response = self.boto3_client.get_bucket_policy(Bucket=bucket)
log_command_execution("S3 get_bucket_policy result", response) log_command_execution(self.s3gate_endpoint, "S3 get_bucket_policy result", response, {"Bucket": bucket})
return response.get("Policy") return response.get("Policy")
@reporter.step("Delete bucket policy") @reporter.step("Delete bucket policy")
@report_error @report_error
def delete_bucket_policy(self, bucket: str) -> str: def delete_bucket_policy(self, bucket: str) -> str:
response = self.boto3_client.delete_bucket_policy(Bucket=bucket) response = self.boto3_client.delete_bucket_policy(Bucket=bucket)
log_command_execution("S3 delete_bucket_policy result", response) log_command_execution(self.s3gate_endpoint, "S3 delete_bucket_policy result", response, {"Bucket": bucket})
return 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:
response = self.boto3_client.put_bucket_policy(Bucket=bucket, Policy=json.dumps(policy)) params = {"Bucket": bucket, "Policy": json.dumps(policy)}
log_command_execution("S3 put_bucket_policy result", response) response = self.boto3_client.put_bucket_policy(**params)
log_command_execution(self.s3gate_endpoint, "S3 put_bucket_policy result", response, params)
return response return response
@reporter.step("Get bucket cors") @reporter.step("Get bucket cors")
@report_error @report_error
def get_bucket_cors(self, bucket: str) -> dict: def get_bucket_cors(self, bucket: str) -> dict:
response = self.boto3_client.get_bucket_cors(Bucket=bucket) response = self.boto3_client.get_bucket_cors(Bucket=bucket)
log_command_execution("S3 get_bucket_cors result", response) log_command_execution(self.s3gate_endpoint, "S3 get_bucket_cors result", response, {"Bucket": bucket})
return response.get("CORSRules") return response.get("CORSRules")
@reporter.step("Get bucket location") @reporter.step("Get bucket location")
@report_error @report_error
def get_bucket_location(self, bucket: str) -> str: def get_bucket_location(self, bucket: str) -> str:
response = self.boto3_client.get_bucket_location(Bucket=bucket) response = self.boto3_client.get_bucket_location(Bucket=bucket)
log_command_execution("S3 get_bucket_location result", response) log_command_execution(self.s3gate_endpoint, "S3 get_bucket_location result", response, {"Bucket": bucket})
return response.get("LocationConstraint") return response.get("LocationConstraint")
@reporter.step("Put bucket cors") @reporter.step("Put bucket cors")
@report_error @report_error
def put_bucket_cors(self, bucket: str, cors_configuration: dict) -> None: def put_bucket_cors(self, bucket: str, cors_configuration: dict) -> None:
response = self.boto3_client.put_bucket_cors(Bucket=bucket, CORSConfiguration=cors_configuration) params = self._convert_to_s3_params(locals().items())
log_command_execution("S3 put_bucket_cors result", response) response = self.boto3_client.put_bucket_cors(**params)
log_command_execution(self.s3gate_endpoint, "S3 put_bucket_cors result", response, params)
return response return response
@reporter.step("Delete bucket cors") @reporter.step("Delete bucket cors")
@report_error @report_error
def delete_bucket_cors(self, bucket: str) -> None: def delete_bucket_cors(self, bucket: str) -> None:
response = self.boto3_client.delete_bucket_cors(Bucket=bucket) response = self.boto3_client.delete_bucket_cors(Bucket=bucket)
log_command_execution("S3 delete_bucket_cors result", response) log_command_execution(self.s3gate_endpoint, "S3 delete_bucket_cors result", response, {"Bucket": bucket})
# END OF BUCKET METHODS # # END OF BUCKET METHODS #
# OBJECT METHODS # # OBJECT METHODS #
@ -284,8 +302,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
@reporter.step("List objects S3 v2") @reporter.step("List objects S3 v2")
@report_error @report_error
def list_objects_v2(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]: def list_objects_v2(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.list_objects_v2(Bucket=bucket) response = self.boto3_client.list_objects_v2(Bucket=bucket)
log_command_execution("S3 v2 List objects result", response) log_command_execution(self.s3gate_endpoint, "S3 v2 List objects result", response, params)
obj_list = [obj["Key"] for obj in response.get("Contents", [])] obj_list = [obj["Key"] for obj in response.get("Contents", [])]
logger.info(f"Found s3 objects: {obj_list}") logger.info(f"Found s3 objects: {obj_list}")
@ -295,8 +314,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
@reporter.step("List objects S3") @reporter.step("List objects S3")
@report_error @report_error
def list_objects(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]: def list_objects(self, bucket: str, full_output: bool = False) -> Union[dict, list[str]]:
params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.list_objects(Bucket=bucket) response = self.boto3_client.list_objects(Bucket=bucket)
log_command_execution("S3 List objects result", response) log_command_execution(self.s3gate_endpoint, "S3 List objects result", response, params)
obj_list = [obj["Key"] for obj in response.get("Contents", [])] obj_list = [obj["Key"] for obj in response.get("Contents", [])]
logger.info(f"Found s3 objects: {obj_list}") logger.info(f"Found s3 objects: {obj_list}")
@ -306,15 +326,17 @@ class Boto3ClientWrapper(S3ClientWrapper):
@reporter.step("List objects versions S3") @reporter.step("List objects versions S3")
@report_error @report_error
def list_objects_versions(self, bucket: str, full_output: bool = False) -> dict: def list_objects_versions(self, bucket: str, full_output: bool = False) -> dict:
params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.list_object_versions(Bucket=bucket) response = self.boto3_client.list_object_versions(Bucket=bucket)
log_command_execution("S3 List objects versions result", response) log_command_execution(self.s3gate_endpoint, "S3 List objects versions result", response, params)
return response if full_output else response.get("Versions", []) return response if full_output else response.get("Versions", [])
@reporter.step("List objects delete markers S3") @reporter.step("List objects delete markers S3")
@report_error @report_error
def list_delete_markers(self, bucket: str, full_output: bool = False) -> list: def list_delete_markers(self, bucket: str, full_output: bool = False) -> list:
params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.list_object_versions(Bucket=bucket) response = self.boto3_client.list_object_versions(Bucket=bucket)
log_command_execution("S3 List objects delete markers result", response) log_command_execution(self.s3gate_endpoint, "S3 List objects delete markers result", response, params)
return response if full_output else response.get("DeleteMarkers", []) return response if full_output else response.get("DeleteMarkers", [])
@reporter.step("Put object S3") @reporter.step("Put object S3")
@ -339,36 +361,33 @@ class Boto3ClientWrapper(S3ClientWrapper):
with open(filepath, "rb") as put_file: with open(filepath, "rb") as put_file:
body = put_file.read() body = put_file.read()
params = { params = self._convert_to_s3_params(locals().items(), exclude=["self", "filepath", "put_file", "body"])
self._to_s3_param(param): value response = self.boto3_client.put_object(Body=body, **params)
for param, value in locals().items() log_command_execution(self.s3gate_endpoint, "S3 Put object result", response, params)
if param not in ["self", "filepath", "put_file"] and value is not None
}
response = self.boto3_client.put_object(**params)
log_command_execution("S3 Put object result", response)
return response.get("VersionId") return response.get("VersionId")
@reporter.step("Head object S3") @reporter.step("Head object S3")
@report_error @report_error
def head_object(self, bucket: str, key: str, version_id: Optional[str] = None) -> dict: def head_object(self, bucket: str, key: str, version_id: Optional[str] = None) -> dict:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.head_object(**params) response = self.boto3_client.head_object(**params)
log_command_execution("S3 Head object result", response) log_command_execution(self.s3gate_endpoint, "S3 Head object result", response, params)
return response return response
@reporter.step("Delete object S3") @reporter.step("Delete object S3")
@report_error @report_error
def delete_object(self, bucket: str, key: str, version_id: Optional[str] = None) -> dict: def delete_object(self, bucket: str, key: str, version_id: Optional[str] = None) -> dict:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.delete_object(**params) response = self.boto3_client.delete_object(**params)
log_command_execution("S3 Delete object result", response) log_command_execution(self.s3gate_endpoint, "S3 Delete object result", response, params)
return response return response
@reporter.step("Delete objects S3") @reporter.step("Delete objects S3")
@report_error @report_error
def delete_objects(self, bucket: str, keys: list[str]) -> dict: def delete_objects(self, bucket: str, keys: list[str]) -> dict:
response = self.boto3_client.delete_objects(Bucket=bucket, Delete=_make_objs_dict(keys)) params = {"Bucket": bucket, "Delete": _make_objs_dict(keys)}
log_command_execution("S3 Delete objects result", response) response = self.boto3_client.delete_objects(**params)
log_command_execution(self.s3gate_endpoint, "S3 Delete objects result", response, params)
assert ( assert (
"Errors" not in response "Errors" not in response
), f'The following objects have not been deleted: {[err_info["Key"] for err_info in response["Errors"]]}.\nError Message: {response["Errors"]["Message"]}' ), f'The following objects have not been deleted: {[err_info["Key"] for err_info in response["Errors"]]}.\nError Message: {response["Errors"]["Message"]}'
@ -387,8 +406,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
for object_version in object_versions for object_version in object_versions
] ]
} }
response = self.boto3_client.delete_objects(Bucket=bucket, Delete=delete_list) params = {"Bucket": bucket, "Delete": delete_list}
log_command_execution("S3 Delete objects result", response) response = self.boto3_client.delete_objects(**params)
log_command_execution(self.s3gate_endpoint, "S3 Delete objects result", response, params)
return response return response
@reporter.step("Delete object versions S3 without delete markers") @reporter.step("Delete object versions S3 without delete markers")
@ -396,8 +416,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
def delete_object_versions_without_dm(self, bucket: str, object_versions: list) -> None: def delete_object_versions_without_dm(self, bucket: str, object_versions: list) -> None:
# Delete objects without creating delete markers # Delete objects without creating delete markers
for object_version in object_versions: for object_version in object_versions:
response = self.boto3_client.delete_object(Bucket=bucket, Key=object_version["Key"], VersionId=object_version["VersionId"]) params = {"Bucket": bucket, "Key": object_version["Key"], "VersionId": object_version["VersionId"]}
log_command_execution("S3 Delete object result", response) response = self.boto3_client.delete_object(**params)
log_command_execution(self.s3gate_endpoint, "S3 Delete object result", response, params)
@reporter.step("Put object ACL") @reporter.step("Put object ACL")
@report_error @report_error
@ -409,17 +430,17 @@ class Boto3ClientWrapper(S3ClientWrapper):
grant_write: Optional[str] = None, grant_write: Optional[str] = None,
grant_read: Optional[str] = None, grant_read: Optional[str] = None,
) -> list: ) -> list:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.put_object_acl(**params) response = self.boto3_client.put_object_acl(**params)
log_command_execution("S3 put object ACL", response) log_command_execution(self.s3gate_endpoint, "S3 put object ACL", response, params)
return response.get("Grants") return response.get("Grants")
@reporter.step("Get object ACL") @reporter.step("Get object ACL")
@report_error @report_error
def get_object_acl(self, bucket: str, key: str, version_id: Optional[str] = None) -> list: def get_object_acl(self, bucket: str, key: str, version_id: Optional[str] = None) -> list:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.get_object_acl(**params) response = self.boto3_client.get_object_acl(**params)
log_command_execution("S3 ACL objects result", response) log_command_execution(self.s3gate_endpoint, "S3 ACL objects result", response, params)
return response.get("Grants") return response.get("Grants")
@reporter.step("Copy object S3") @reporter.step("Copy object S3")
@ -442,13 +463,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
key = string_utils.unique_name("copy-object-") key = string_utils.unique_name("copy-object-")
copy_source = f"{source_bucket}/{source_key}" copy_source = f"{source_bucket}/{source_key}"
params = { params = self._convert_to_s3_params(locals().items(), exclude=["self", "source_bucket", "source_key"])
self._to_s3_param(param): value
for param, value in locals().items()
if param not in ["self", "source_bucket", "source_key"] and value is not None
}
response = self.boto3_client.copy_object(**params) response = self.boto3_client.copy_object(**params)
log_command_execution("S3 Copy objects result", response) log_command_execution(self.s3gate_endpoint, "S3 Copy objects result", response, params)
return key return key
@reporter.step("Get object S3") @reporter.step("Get object S3")
@ -465,13 +482,12 @@ class Boto3ClientWrapper(S3ClientWrapper):
if object_range: if object_range:
range_str = f"bytes={object_range[0]}-{object_range[1]}" range_str = f"bytes={object_range[0]}-{object_range[1]}"
params = { params = self._convert_to_s3_params(
self._to_s3_param(param): value {**locals(), **{"Range": range_str}}.items(),
for param, value in {**locals(), **{"Range": range_str}}.items() exclude=["self", "object_range", "full_output", "range_str"],
if param not in ["self", "object_range", "full_output", "range_str", "filename"] and value is not None )
}
response = self.boto3_client.get_object(**params) response = self.boto3_client.get_object(**params)
log_command_execution("S3 Get objects result", response) log_command_execution(self.s3gate_endpoint, "S3 Get objects result", response, params)
if full_output: if full_output:
return response return response
@ -487,8 +503,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
@reporter.step("Create multipart upload S3") @reporter.step("Create multipart upload S3")
@report_error @report_error
def create_multipart_upload(self, bucket: str, key: str) -> str: def create_multipart_upload(self, bucket: str, key: str) -> str:
response = self.boto3_client.create_multipart_upload(Bucket=bucket, Key=key) params = self._convert_to_s3_params(locals().items())
log_command_execution("S3 Created multipart upload", response) response = self.boto3_client.create_multipart_upload(**params)
log_command_execution(self.s3gate_endpoint, "S3 Created multipart upload", response, params)
assert response.get("UploadId"), f"Expected UploadId in response:\n{response}" assert response.get("UploadId"), f"Expected UploadId in response:\n{response}"
return response["UploadId"] return response["UploadId"]
@ -497,15 +514,16 @@ class Boto3ClientWrapper(S3ClientWrapper):
@report_error @report_error
def list_multipart_uploads(self, bucket: str) -> Optional[list[dict]]: def list_multipart_uploads(self, bucket: str) -> Optional[list[dict]]:
response = self.boto3_client.list_multipart_uploads(Bucket=bucket) response = self.boto3_client.list_multipart_uploads(Bucket=bucket)
log_command_execution("S3 List multipart upload", response) log_command_execution(self.s3gate_endpoint, "S3 List multipart upload", response, {"Bucket": bucket})
return response.get("Uploads") return response.get("Uploads")
@reporter.step("Abort multipart upload S3") @reporter.step("Abort multipart upload S3")
@report_error @report_error
def abort_multipart_upload(self, bucket: str, key: str, upload_id: str) -> None: def abort_multipart_upload(self, bucket: str, key: str, upload_id: str) -> None:
response = self.boto3_client.abort_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id) params = self._convert_to_s3_params(locals().items())
log_command_execution("S3 Abort multipart upload", response) response = self.boto3_client.abort_multipart_upload(**params)
log_command_execution(self.s3gate_endpoint, "S3 Abort multipart upload", response, params)
@reporter.step("Upload part S3") @reporter.step("Upload part S3")
@report_error @report_error
@ -513,14 +531,10 @@ class Boto3ClientWrapper(S3ClientWrapper):
with open(filepath, "rb") as put_file: with open(filepath, "rb") as put_file:
body = put_file.read() body = put_file.read()
response = self.boto3_client.upload_part( params = self._convert_to_s3_params(locals().items(), exclude=["self", "put_file", "part_num", "filepath", "body"])
UploadId=upload_id, params["PartNumber"] = part_num
Bucket=bucket, response = self.boto3_client.upload_part(Body=body, **params)
Key=key, log_command_execution(self.s3gate_endpoint, "S3 Upload part", response, params)
PartNumber=part_num,
Body=body,
)
log_command_execution("S3 Upload part", response)
assert response.get("ETag"), f"Expected ETag in response:\n{response}" assert response.get("ETag"), f"Expected ETag in response:\n{response}"
return response["ETag"] return response["ETag"]
@ -528,14 +542,10 @@ class Boto3ClientWrapper(S3ClientWrapper):
@reporter.step("Upload copy part S3") @reporter.step("Upload copy part S3")
@report_error @report_error
def upload_part_copy(self, bucket: str, key: str, upload_id: str, part_num: int, copy_source: str) -> str: def upload_part_copy(self, bucket: str, key: str, upload_id: str, part_num: int, copy_source: str) -> str:
response = self.boto3_client.upload_part_copy( params = self._convert_to_s3_params(locals().items(), exclude=["self", "put_file", "part_num", "filepath"])
UploadId=upload_id, params["PartNumber"] = part_num
Bucket=bucket, response = self.boto3_client.upload_part_copy(**params)
Key=key, log_command_execution(self.s3gate_endpoint, "S3 Upload copy part", response, params)
PartNumber=part_num,
CopySource=copy_source,
)
log_command_execution("S3 Upload copy part", response)
assert response.get("CopyPartResult", []).get("ETag"), f"Expected ETag in response:\n{response}" assert response.get("CopyPartResult", []).get("ETag"), f"Expected ETag in response:\n{response}"
return response["CopyPartResult"]["ETag"] return response["CopyPartResult"]["ETag"]
@ -543,8 +553,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
@reporter.step("List parts S3") @reporter.step("List parts S3")
@report_error @report_error
def list_parts(self, bucket: str, key: str, upload_id: str) -> list[dict]: def list_parts(self, bucket: str, key: str, upload_id: str) -> list[dict]:
response = self.boto3_client.list_parts(UploadId=upload_id, Bucket=bucket, Key=key) params = self._convert_to_s3_params(locals().items())
log_command_execution("S3 List part", response) response = self.boto3_client.list_parts(**params)
log_command_execution(self.s3gate_endpoint, "S3 List part", response, params)
assert response.get("Parts"), f"Expected Parts in response:\n{response}" assert response.get("Parts"), f"Expected Parts in response:\n{response}"
return response["Parts"] return response["Parts"]
@ -553,8 +564,10 @@ class Boto3ClientWrapper(S3ClientWrapper):
@report_error @report_error
def complete_multipart_upload(self, bucket: str, key: str, upload_id: str, parts: list) -> None: def complete_multipart_upload(self, bucket: str, key: str, upload_id: str, parts: list) -> None:
parts = [{"ETag": etag, "PartNumber": part_num} for part_num, etag in parts] parts = [{"ETag": etag, "PartNumber": part_num} for part_num, etag in parts]
response = self.boto3_client.complete_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id, MultipartUpload={"Parts": parts}) params = self._convert_to_s3_params(locals().items(), exclude=["self", "parts"])
log_command_execution("S3 Complete multipart upload", response) params["MultipartUpload"] = {"Parts": parts}
response = self.boto3_client.complete_multipart_upload(**params)
log_command_execution(self.s3gate_endpoint, "S3 Complete multipart upload", response, params)
return response return response
@ -568,9 +581,9 @@ class Boto3ClientWrapper(S3ClientWrapper):
version_id: Optional[str] = None, version_id: Optional[str] = None,
bypass_governance_retention: Optional[bool] = None, bypass_governance_retention: Optional[bool] = None,
) -> None: ) -> None:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.put_object_retention(**params) response = self.boto3_client.put_object_retention(**params)
log_command_execution("S3 Put object retention ", response) log_command_execution(self.s3gate_endpoint, "S3 Put object retention ", response, params)
@reporter.step("Put object legal hold") @reporter.step("Put object legal hold")
@report_error @report_error
@ -582,35 +595,33 @@ class Boto3ClientWrapper(S3ClientWrapper):
version_id: Optional[str] = None, version_id: Optional[str] = None,
) -> None: ) -> None:
legal_hold = {"Status": legal_hold_status} legal_hold = {"Status": legal_hold_status}
params = { params = self._convert_to_s3_params(locals().items(), exclude=["self", "legal_hold_status"])
self._to_s3_param(param): value
for param, value in locals().items()
if param not in ["self", "legal_hold_status"] and value is not None
}
response = self.boto3_client.put_object_legal_hold(**params) response = self.boto3_client.put_object_legal_hold(**params)
log_command_execution("S3 Put object legal hold ", response) log_command_execution(self.s3gate_endpoint, "S3 Put object legal hold ", response, params)
@reporter.step("Put object tagging") @reporter.step("Put object tagging")
@report_error @report_error
def put_object_tagging(self, bucket: str, key: str, tags: list, version_id: Optional[str] = "") -> None: def put_object_tagging(self, bucket: str, key: str, tags: list, version_id: Optional[str] = "") -> None:
tags = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags] tags = [{"Key": tag_key, "Value": tag_value} for tag_key, tag_value in tags]
tagging = {"TagSet": tags} tagging = {"TagSet": tags}
response = self.boto3_client.put_object_tagging(Bucket=bucket, Key=key, Tagging=tagging, VersionId=version_id) params = self._convert_to_s3_params(locals().items(), exclude=["self", "tags"])
log_command_execution("S3 Put object tagging", response) response = self.boto3_client.put_object_tagging(**params)
log_command_execution(self.s3gate_endpoint, "S3 Put object tagging", response, params)
@reporter.step("Get object tagging") @reporter.step("Get object tagging")
@report_error @report_error
def get_object_tagging(self, bucket: str, key: str, version_id: Optional[str] = None) -> list: def get_object_tagging(self, bucket: str, key: str, version_id: Optional[str] = None) -> list:
params = {self._to_s3_param(param): value for param, value in locals().items() if param not in ["self"] and value is not None} params = self._convert_to_s3_params(locals().items())
response = self.boto3_client.get_object_tagging(**params) response = self.boto3_client.get_object_tagging(**params)
log_command_execution("S3 Get object tagging", response) log_command_execution(self.s3gate_endpoint, "S3 Get object tagging", response, params)
return response.get("TagSet") return response.get("TagSet")
@reporter.step("Delete object tagging") @reporter.step("Delete object tagging")
@report_error @report_error
def delete_object_tagging(self, bucket: str, key: str) -> None: def delete_object_tagging(self, bucket: str, key: str) -> None:
response = self.boto3_client.delete_object_tagging(Bucket=bucket, Key=key) params = self._convert_to_s3_params(locals().items())
log_command_execution("S3 Delete object tagging", response) response = self.boto3_client.delete_object_tagging(**params)
log_command_execution(self.s3gate_endpoint, "S3 Delete object tagging", response, params)
@reporter.step("Get object attributes") @reporter.step("Get object attributes")
@report_error @report_error

10
src/frostfs_testlib/utils/cli_utils.py
View file

@ -75,12 +75,18 @@ def _attach_allure_log(cmd: str, output: str, return_code: int, start_time: date
reporter.attach(command_attachment, "Command execution") reporter.attach(command_attachment, "Command execution")
def log_command_execution(cmd: str, output: Union[str, TypedDict]) -> None: def log_command_execution(url: str, cmd: str, output: Union[str, TypedDict], params: dict = None) -> None:
logger.info(f"{cmd}: {output}") logger.info(f"{cmd}: {output}")
with suppress(Exception): with suppress(Exception):
json_output = json.dumps(output, indent=4, sort_keys=True) json_output = json.dumps(output, indent=4, sort_keys=True)
output = json_output output = json_output
command_attachment = f"COMMAND: '{cmd}'\n" f"OUTPUT:\n {output}\n"
with suppress(Exception):
json_params = json.dumps(params, indent=4, sort_keys=True)
params = json_params
command_attachment = f"COMMAND: '{cmd}'\n" f"URL: {url}\n" f"PARAMS:\n{params}\n" f"OUTPUT:\n{output}\n"
with reporter.step(f'COMMAND: {shorten(cmd, width=60, placeholder="...")}'): with reporter.step(f'COMMAND: {shorten(cmd, width=60, placeholder="...")}'):
reporter.attach(command_attachment, "Command execution") reporter.attach(command_attachment, "Command execution")