ChEktn/frostfs-testcases
1
0
Fork 0
forked from TrueCloudLab/frostfs-testcases
Code Pull requests Activity

Disable automatic pagination in aws cli client

Browse source 
This should prevent output truncation if response contains too many items.

Signed-off-by: Vladimir Domnich <v.domnich@yadro.com>
This commit is contained in:
Vladimir Domnich 2022-07-26 17:04:00 +03:00
parent d8911f2490
commit 8a7d8f7c39
1 changed files with 66 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

119
robot/resources/lib/python_keywords/aws_cli_client.py
View file

@ -10,6 +10,9 @@ logger = logging.getLogger('NeoLogger')
class AwsCliClient: class AwsCliClient:
# 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"
def __init__(self, access_key_id: str, secret_access_key: str): def __init__(self, access_key_id: str, secret_access_key: str):
self.access_key_id = access_key_id self.access_key_id = access_key_id
@ -22,73 +25,74 @@ class AwsCliClient:
_configure_aws_cli(cmd, self.access_key_id, self.secret_access_key) _configure_aws_cli(cmd, self.access_key_id, self.secret_access_key)
def create_bucket(self, Bucket: str): def create_bucket(self, Bucket: str):
cmd = f'aws --no-verify-ssl s3api create-bucket --bucket {Bucket} --endpoint-url {S3_GATE}' cmd = f'aws {self.common_flags} s3api create-bucket --bucket {Bucket} ' \
f'--endpoint {S3_GATE}'
_cmd_run(cmd, timeout=90) _cmd_run(cmd, timeout=90)
def list_buckets(self) -> dict: def list_buckets(self) -> dict:
cmd = f'aws --no-verify-ssl s3api list-buckets --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api list-buckets --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def get_bucket_versioning(self, Bucket: str) -> dict: def get_bucket_versioning(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api get-bucket-versioning --bucket {Bucket}' \ cmd = f'aws {self.common_flags} s3api get-bucket-versioning --bucket {Bucket} ' \
f' --endpoint {S3_GATE}' f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def put_bucket_versioning(self, Bucket: str, VersioningConfiguration: dict) -> dict: def put_bucket_versioning(self, Bucket: str, VersioningConfiguration: dict) -> dict:
cmd = f'aws --no-verify-ssl s3api put-bucket-versioning --bucket {Bucket} ' \ cmd = f'aws {self.common_flags} s3api put-bucket-versioning --bucket {Bucket} ' \
f'--versioning-configuration Status={VersioningConfiguration.get("Status")}' \ f'--versioning-configuration Status={VersioningConfiguration.get("Status")} ' \
f' --endpoint {S3_GATE}' f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def list_objects(self, Bucket: str) -> dict: def list_objects(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api list-objects --bucket {Bucket}' \ cmd = f'aws {self.common_flags} s3api list-objects --bucket {Bucket} ' \
f' --endpoint {S3_GATE}' f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def list_objects_v2(self, Bucket: str) -> dict: def list_objects_v2(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api list-objects-v2 --bucket {Bucket}' \ cmd = f'aws {self.common_flags} s3api list-objects-v2 --bucket {Bucket} ' \
f' --endpoint {S3_GATE}' f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def list_object_versions(self, Bucket: str) -> dict: def list_object_versions(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api list-object-versions --bucket {Bucket}' \ cmd = f'aws {self.common_flags} s3api list-object-versions --bucket {Bucket} ' \
f' --endpoint {S3_GATE}' f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def copy_object(self, Bucket: str, CopySource: str, Key: str) -> dict: def copy_object(self, Bucket: str, CopySource: str, Key: str) -> dict:
cmd = f'aws --no-verify-ssl s3api copy-object --copy-source {CopySource} --bucket {Bucket} --key {Key}' \ cmd = f'aws {self.common_flags} s3api copy-object --copy-source {CopySource} ' \
f' --endpoint {S3_GATE}' f'--bucket {Bucket} --key {Key} --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def head_bucket(self, Bucket: str) -> dict: def head_bucket(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api head-bucket --bucket {Bucket} --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api head-bucket --bucket {Bucket} --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def put_object(self, Body: str, Bucket: str, Key: str) -> dict: def put_object(self, Body: str, Bucket: str, Key: str) -> dict:
cmd = f' aws --no-verify-ssl s3api put-object --bucket {Bucket} --key {Key} --body {Body}' \ cmd = f'aws {self.common_flags} s3api put-object --bucket {Bucket} --key {Key} ' \
f' --endpoint {S3_GATE}' f'--body {Body} --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def head_object(self, Bucket: str, Key: str, VersionId: str = None) -> dict: def head_object(self, Bucket: str, Key: str, VersionId: str = None) -> dict:
version = f' --version-id {VersionId}' if VersionId else '' version = f' --version-id {VersionId}' if VersionId else ''
cmd = f' aws --no-verify-ssl s3api head-object --bucket {Bucket} --key {Key} {version}' \ cmd = f'aws {self.common_flags} s3api head-object --bucket {Bucket} --key {Key} ' \
f' --endpoint {S3_GATE}' f'{version} --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def get_object(self, Bucket: str, Key: str, file_path: str, VersionId: str = None) -> dict: def get_object(self, Bucket: str, Key: str, file_path: str, VersionId: str = None) -> dict:
version = f' --version-id {VersionId}' if VersionId else '' version = f' --version-id {VersionId}' if VersionId else ''
cmd = f' aws --no-verify-ssl s3api get-object --bucket {Bucket} ' \ cmd = f'aws {self.common_flags} s3api get-object --bucket {Bucket} --key {Key} ' \
f'--key {Key} {version} {file_path} --endpoint {S3_GATE}' f'{version} {file_path} --endpoint {S3_GATE}'
output = _cmd_run(cmd, timeout=90) output = _cmd_run(cmd, timeout=90)
return self._to_json(output) return self._to_json(output)
@ -96,16 +100,17 @@ class AwsCliClient:
file_path = f"{os.getcwd()}/{ASSETS_DIR}/delete.json" file_path = f"{os.getcwd()}/{ASSETS_DIR}/delete.json"
with open(file_path, 'w') as out_file: with open(file_path, 'w') as out_file:
out_file.write(json.dumps(Delete)) out_file.write(json.dumps(Delete))
logger.info(f"Input file for delete-objects: {json.dumps(Delete)}")
cmd = f'aws --no-verify-ssl s3api delete-objects --bucket {Bucket} --delete file://{file_path} ' \ cmd = f'aws {self.common_flags} s3api delete-objects --bucket {Bucket} ' \
f'--endpoint {S3_GATE}' f'--delete file://{file_path} --endpoint {S3_GATE}'
output = _cmd_run(cmd, timeout=90) output = _cmd_run(cmd, timeout=90)
return self._to_json(output) return self._to_json(output)
def delete_object(self, Bucket: str, Key: str, VersionId: str = None) -> dict: def delete_object(self, Bucket: str, Key: str, VersionId: str = None) -> dict:
version = f' --version-id {VersionId}' if VersionId else '' version = f' --version-id {VersionId}' if VersionId else ''
cmd = f'aws --no-verify-ssl s3api delete-object --bucket {Bucket} --key {Key} {version}' \ cmd = f'aws {self.common_flags} s3api delete-object --bucket {Bucket} ' \
f' --endpoint {S3_GATE}' f'--key {Key} {version} --endpoint {S3_GATE}'
output = _cmd_run(cmd, timeout=90) output = _cmd_run(cmd, timeout=90)
return self._to_json(output) return self._to_json(output)
@ -115,93 +120,101 @@ class AwsCliClient:
version = f' --version-id {version_id}' if version_id else '' version = f' --version-id {version_id}' if version_id else ''
parts = f'--max-parts {max_parts}' if max_parts else '' parts = f'--max-parts {max_parts}' if max_parts else ''
part_number = f'--part-number-marker {part_number}' if part_number else '' part_number = f'--part-number-marker {part_number}' if part_number else ''
cmd = f'aws --no-verify-ssl s3api get-object-attributes --bucket {bucket} --key {key} {version}' \ cmd = f'aws {self.common_flags} s3api get-object-attributes --bucket {bucket} ' \
f' {parts} {part_number} --object-attributes {attrs} --endpoint {S3_GATE}' f'--key {key} {version} {parts} {part_number} --object-attributes {attrs} ' \
f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def delete_bucket(self, Bucket: str) -> dict: def delete_bucket(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api delete-bucket --bucket {Bucket} --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api delete-bucket --bucket {Bucket} --endpoint {S3_GATE}'
output = _cmd_run(cmd, timeout=90) output = _cmd_run(cmd, timeout=90)
return self._to_json(output) return self._to_json(output)
def get_bucket_tagging(self, Bucket: str) -> dict: def get_bucket_tagging(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api get-bucket-tagging --bucket {Bucket} --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api get-bucket-tagging --bucket {Bucket} ' \
f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def put_bucket_tagging(self, Bucket: str, Tagging: dict) -> dict: def put_bucket_tagging(self, Bucket: str, Tagging: dict) -> dict:
cmd = f'aws --no-verify-ssl s3api put-bucket-tagging --bucket {Bucket} --tagging \'{json.dumps(Tagging)}\'' \ cmd = f"aws {self.common_flags} s3api put-bucket-tagging --bucket {Bucket} " \
f' --endpoint {S3_GATE}' f"--tagging '{json.dumps(Tagging)}' --endpoint {S3_GATE}"
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def delete_bucket_tagging(self, Bucket: str) -> dict: def delete_bucket_tagging(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api delete-bucket-tagging --bucket {Bucket} --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api delete-bucket-tagging --bucket {Bucket} ' \
f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def put_object_tagging(self, Bucket: str, Key: str, Tagging: dict) -> dict: def put_object_tagging(self, Bucket: str, Key: str, Tagging: dict) -> dict:
cmd = f'aws --no-verify-ssl s3api put-object-tagging --bucket {Bucket} --key {Key}' \ cmd = f"aws {self.common_flags} s3api put-object-tagging --bucket {Bucket} --key {Key} " \
f' --tagging \'{json.dumps(Tagging)}\'' \ f"--tagging '{json.dumps(Tagging)}' --endpoint {S3_GATE}"
f' --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def get_object_tagging(self, Bucket: str, Key: str) -> dict: def get_object_tagging(self, Bucket: str, Key: str) -> dict:
cmd = f'aws --no-verify-ssl s3api get-object-tagging --bucket {Bucket} --key {Key} --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api get-object-tagging --bucket {Bucket} --key {Key} ' \
f'--endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def delete_object_tagging(self, Bucket: str, Key: str) -> dict: def delete_object_tagging(self, Bucket: str, Key: str) -> dict:
cmd = f'aws --no-verify-ssl s3api delete-object-tagging --bucket {Bucket} --key {Key} --endpoint {S3_GATE}' cmd = f'aws {self.common_flags} s3api delete-object-tagging --bucket {Bucket} ' \
f'--key {Key} --endpoint {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
@allure.step('Sync directory S3') @allure.step('Sync directory S3')
def sync(self, bucket_name: str, dir_path: str) -> dict: def sync(self, bucket_name: str, dir_path: str) -> dict:
cmd = f'aws --no-verify-ssl s3 sync {dir_path} s3://{bucket_name} --endpoint-url {S3_GATE}' cmd = f'aws {self.common_flags} s3 sync {dir_path} s3://{bucket_name} ' \
f'--endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def create_multipart_upload(self, Bucket: str, Key: str) -> dict: def create_multipart_upload(self, Bucket: str, Key: str) -> dict:
cmd = f'aws --no-verify-ssl s3api create-multipart-upload --bucket {Bucket} --key {Key}' \ cmd = f'aws {self.common_flags} s3api create-multipart-upload --bucket {Bucket} ' \
f' --endpoint-url {S3_GATE}' f'--key {Key} --endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def list_multipart_uploads(self, Bucket: str) -> dict: def list_multipart_uploads(self, Bucket: str) -> dict:
cmd = f'aws --no-verify-ssl s3api list-multipart-uploads --bucket {Bucket}' \ cmd = f'aws {self.common_flags} s3api list-multipart-uploads --bucket {Bucket} ' \
f' --endpoint-url {S3_GATE}' f'--endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def abort_multipart_upload(self, Bucket: str, Key: str, UploadId: str) -> dict: def abort_multipart_upload(self, Bucket: str, Key: str, UploadId: str) -> dict:
cmd = f'aws --no-verify-ssl s3api abort-multipart-upload --bucket {Bucket} --key {Key}' \ cmd = f'aws {self.common_flags} s3api abort-multipart-upload --bucket {Bucket} ' \
f' --upload-id {UploadId} --endpoint-url {S3_GATE}' f'--key {Key} --upload-id {UploadId} --endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def upload_part(self, UploadId: str, Bucket: str, Key: str, PartNumber: int, Body: str) -> dict: def upload_part(self, UploadId: str, Bucket: str, Key: str, PartNumber: int, Body: str) -> dict:
cmd = f'aws --no-verify-ssl s3api upload-part --bucket {Bucket} --key {Key} --upload-id {UploadId} ' \ cmd = f'aws {self.common_flags} s3api upload-part --bucket {Bucket} --key {Key} ' \
f'--part-number {PartNumber} --body {Body} --endpoint-url {S3_GATE}' f'--upload-id {UploadId} --part-number {PartNumber} --body {Body} ' \
f'--endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def list_parts(self, UploadId: str, Bucket: str, Key: str) -> dict: def list_parts(self, UploadId: str, Bucket: str, Key: str) -> dict:
cmd = f'aws --no-verify-ssl s3api list-parts --bucket {Bucket} --key {Key} --upload-id {UploadId} ' \ cmd = f'aws {self.common_flags} s3api list-parts --bucket {Bucket} --key {Key} ' \
f' --endpoint-url {S3_GATE}' f'--upload-id {UploadId} --endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)
def complete_multipart_upload(self, Bucket: str, Key: str, UploadId: str, MultipartUpload: dict) -> dict: def complete_multipart_upload(self, Bucket: str, Key: str, UploadId: str,
MultipartUpload: dict) -> dict:
file_path = f"{os.getcwd()}/{ASSETS_DIR}/parts.json" file_path = f"{os.getcwd()}/{ASSETS_DIR}/parts.json"
with open(file_path, 'w') as out_file: with open(file_path, 'w') as out_file:
out_file.write(json.dumps(MultipartUpload)) out_file.write(json.dumps(MultipartUpload))
logger.info(f"Input file for complete-multipart-upload: {json.dumps(MultipartUpload)}")
cmd = f'aws --no-verify-ssl s3api complete-multipart-upload --bucket {Bucket} --key {Key}' \ cmd = f'aws {self.common_flags} s3api complete-multipart-upload --bucket {Bucket} ' \
f' --upload-id {UploadId} --multipart-upload file://{file_path}' \ f'--key {Key} --upload-id {UploadId} --multipart-upload file://{file_path} ' \
f' --endpoint-url {S3_GATE}' f'--endpoint-url {S3_GATE}'
output = _cmd_run(cmd) output = _cmd_run(cmd)
return self._to_json(output) return self._to_json(output)