[#223] Add multipart test cases with bucket without versioning #223
1 changed files with 16 additions and 6 deletions
|
@ -16,9 +16,11 @@ PART_SIZE = 5 * 1024 * 1024
|
|||
class TestS3GateMultipart(ClusterTestBase):
|
||||
NO_SUCH_UPLOAD = "The upload ID may be invalid, or the upload may have been aborted or completed."
|
||||
|
||||
@allure.title("Object Multipart API (s3_client={s3_client})")
|
||||
@pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED], indirect=True)
|
||||
def test_s3_object_multipart(self, s3_client: S3ClientWrapper, bucket: str, default_wallet: str):
|
||||
@allure.title("Object Multipart API (s3_client={s3_client}, bucket versioning = {versioning_status})")
|
||||
@pytest.mark.parametrize("versioning_status", [VersioningStatus.ENABLED, VersioningStatus.UNDEFINED], indirect=True)
|
||||
def test_s3_object_multipart(
|
||||
self, s3_client: S3ClientWrapper, bucket: str, default_wallet: str, versioning_status: str
|
||||
):
|
||||
parts_count = 5
|
||||
file_name_large = generate_file(PART_SIZE * parts_count) # 5Mb - min part
|
||||
object_key = s3_helper.object_key_from_file_path(file_name_large)
|
||||
|
@ -44,7 +46,11 @@ class TestS3GateMultipart(ClusterTestBase):
|
|||
etag = s3_client.upload_part(bucket, object_key, upload_id, part_id, file_path)
|
||||
parts.append((part_id, etag))
|
||||
got_parts = s3_client.list_parts(bucket, object_key, upload_id)
|
||||
s3_client.complete_multipart_upload(bucket, object_key, upload_id, parts)
|
||||
response = s3_client.complete_multipart_upload(bucket, object_key, upload_id, parts)
|
||||
|
||||
version_id = None
|
||||
if versioning_status == VersioningStatus.ENABLED:
|
||||
version_id = response["VersionId"]
|
||||
assert len(got_parts) == len(part_files), f"Expected {parts_count} parts, got\n{got_parts}"
|
||||
|
||||
with reporter.step("Check upload list is empty"):
|
||||
|
@ -55,8 +61,12 @@ class TestS3GateMultipart(ClusterTestBase):
|
|||
got_object = s3_client.get_object(bucket, object_key)
|
||||
assert get_file_hash(got_object) == get_file_hash(file_name_large)
|
||||
|
||||
with reporter.step("Delete the object"):
|
||||
s3_client.delete_object(bucket, object_key)
|
||||
if version_id:
|
||||
with reporter.step("Delete the object version"):
|
||||
s3_client.delete_object(bucket, object_key, version_id)
|
||||
else:
|
||||
with reporter.step("Delete the object"):
|
||||
s3_client.delete_object(bucket, object_key)
|
||||
|
||||
with reporter.step("List objects in the bucket, expect to be empty"):
|
||||
objects_list = s3_client.list_objects(bucket)
|
||||
|
|
Loading…
Reference in a new issue