KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use create_multipart_upload method in localstack

Best Python code snippet using localstack_python

test_models.py

Source:test_models.py

...25    upload.complete_multipart_upload(26        parts=[{"ETag": response.headers["ETag"], "PartNumber": 0}]27    )28    assert upload.status == UserUpload.StatusChoices.COMPLETED29def test_create_multipart_upload():30    user = UserFactory.build()31    upload = UserUpload(creator=user)32    assert upload.s3_upload_id == ""33    assert upload.status == UserUpload.StatusChoices.PENDING34    upload.create_multipart_upload()35    assert upload.s3_upload_id != ""36    assert upload.status == UserUpload.StatusChoices.INITIALIZED37    assert upload.key == f"uploads/{user.pk}/{upload.pk}"38def test_generate_presigned_urls():39    upload = UserUpload(creator=UserFactory.build())40    upload.create_multipart_upload()41    presigned_urls = upload.generate_presigned_urls(part_numbers=[1, 13, 26])42    assert set(presigned_urls.keys()) == {"1", "13", "26"}43    assert presigned_urls["1"].startswith(44        f"{settings.AWS_S3_ENDPOINT_URL}/{upload.bucket}/{upload.key}?uploadId={upload.s3_upload_id}&partNumber=1&"45    )46    assert presigned_urls["13"].startswith(47        f"{settings.AWS_S3_ENDPOINT_URL}/{upload.bucket}/{upload.key}?uploadId={upload.s3_upload_id}&partNumber=13&"48    )49    assert presigned_urls["26"].startswith(50        f"{settings.AWS_S3_ENDPOINT_URL}/{upload.bucket}/{upload.key}?uploadId={upload.s3_upload_id}&partNumber=26&"51    )52def test_abort_multipart_upload():53    upload = UserUpload(creator=UserFactory.build())54    upload.create_multipart_upload()55    assert upload.status == UserUpload.StatusChoices.INITIALIZED56    assert upload.s3_upload_id != ""57    upload.abort_multipart_upload()58    assert upload.status == UserUpload.StatusChoices.ABORTED59    assert upload.s3_upload_id == ""60def test_list_parts():61    upload = UserUpload(creator=UserFactory.build())62    upload.create_multipart_upload()63    url = upload.generate_presigned_url(part_number=1)64    response = put(url, data=b"123")65    parts = upload.list_parts()66    assert len(parts) == 167    assert parts[0]["ETag"] == response.headers["ETag"]68    assert parts[0]["Size"] == 369    assert parts[0]["PartNumber"] == 170def test_list_parts_empty():71    upload = UserUpload(creator=UserFactory.build())72    upload.create_multipart_upload()73    parts = upload.list_parts()74    assert parts == []75def test_list_parts_truncation():76    upload = UserUpload(creator=UserFactory.build())77    upload.create_multipart_upload()78    presigned_urls = upload.generate_presigned_urls(part_numbers=[1, 2])79    responses = {}80    for part_number, url in presigned_urls.items():81        responses[part_number] = put(url, data=b"123")82    upload.LIST_MAX_ITEMS = 183    parts = upload.list_parts()84    assert len(parts) == 285    assert parts[0]["ETag"] == responses["1"].headers["ETag"]86    assert parts[0]["Size"] == 387    assert parts[0]["PartNumber"] == 188    assert parts[1]["ETag"] == responses["2"].headers["ETag"]89    assert parts[1]["Size"] == 390    assert parts[1]["PartNumber"] == 291@pytest.mark.django_db92def test_upload_copy():93    user = UserFactory()94    upload = UserUpload.objects.create(creator=user, filename="test.tar.gz")95    presigned_urls = upload.generate_presigned_urls(part_numbers=[1])96    response = put(presigned_urls["1"], data=b"123")97    upload.complete_multipart_upload(98        parts=[{"ETag": response.headers["ETag"], "PartNumber": 1}]99    )100    upload.save()101    ai = AlgorithmImageFactory(creator=user, image=None)102    assert not ai.image103    upload.copy_object(to_field=ai.image)104    assert (105        ai.image.name106        == f"docker/images/algorithms/algorithmimage/{ai.pk}/test.tar.gz"107    )108    assert ai.image.storage.exists(name=ai.image.name)109    with ai.image.open() as f:110        assert f.read() == b"123"111@pytest.mark.django_db112def test_file_deleted_with_object():113    u = UserFactory()114    upload = UserUpload.objects.create(creator=u)115    presigned_urls = upload.generate_presigned_urls(part_numbers=[1])116    response = put(presigned_urls["1"], data=b"123")117    upload.complete_multipart_upload(118        parts=[{"ETag": response.headers["ETag"], "PartNumber": 1}]119    )120    upload.save()121    bucket = upload.bucket122    key = upload.key123    assert upload._client.head_object(Bucket=bucket, Key=key)124    UserUpload.objects.filter(pk=upload.pk).delete()125    with pytest.raises(upload._client.exceptions.ClientError):126        upload._client.head_object(Bucket=bucket, Key=key)127@pytest.mark.django_db128def test_incomplete_deleted_with_object():129    u = UserFactory()130    upload = UserUpload.objects.create(creator=u)131    bucket = upload.bucket132    key = upload.key133    assert "Uploads" in upload._client.list_multipart_uploads(134        Bucket=bucket, Prefix=key135    )136    UserUpload.objects.filter(pk=upload.pk).delete()137    assert "Uploads" not in upload._client.list_multipart_uploads(138        Bucket=bucket, Prefix=key139    )140def test_size_of_creators_completed_uploads():141    def upload_files_for_user(user, n=1):142        for _ in range(n):143            ul = UserUpload(creator=user)144            ul.create_multipart_upload()145            presigned_urls = ul.generate_presigned_urls(part_numbers=[1])146            response = put(presigned_urls["1"], data=b"123")147            ul.complete_multipart_upload(148                parts=[{"ETag": response.headers["ETag"], "PartNumber": 1}]149            )150    u = UserFactory.build(pk=42)151    upload = UserUpload(creator=u)152    upload.LIST_MAX_ITEMS = 1153    initial_upload_size = upload.size_of_creators_completed_uploads154    assert type(initial_upload_size) == int155    upload_files_for_user(user=u, n=upload.LIST_MAX_ITEMS + 1)156    # another users files should not be considered157    upload_files_for_user(user=UserFactory.build(pk=u.pk + 1))158    assert (159        upload.size_of_creators_completed_uploads160        == initial_upload_size + (upload.LIST_MAX_ITEMS + 1) * 3161    )162def test_size_incomplete():163    u = UserFactory.build(pk=42)164    upload = UserUpload(creator=u)165    upload.create_multipart_upload()166    upload.LIST_MAX_ITEMS = 1167    assert upload.size == 0168    parts = [1, 2]169    presigned_urls = upload.generate_presigned_urls(part_numbers=parts)170    for part in parts:171        put(presigned_urls[str(part)], data=b"123")172    assert upload.size == (upload.LIST_MAX_ITEMS + 1) * 3173def test_size_complete():174    u = UserFactory.build(pk=42)175    upload = UserUpload(creator=u)176    upload.create_multipart_upload()177    assert upload.size == 0178    presigned_urls = upload.generate_presigned_urls(part_numbers=[1])179    response = put(presigned_urls["1"], data=b"123")180    upload.complete_multipart_upload(181        parts=[{"ETag": response.headers["ETag"], "PartNumber": 1}]182    )183    assert upload.size == 3184@pytest.mark.django_db185def test_can_upload_more_unverified(settings):186    upload = UserUpload.objects.create(creator=UserFactory())187    presigned_urls = upload.generate_presigned_urls(part_numbers=[1])188    put(presigned_urls["1"], data=b"123")189    assert upload.can_upload_more is True190    settings.UPLOADS_MAX_SIZE_UNVERIFIED = 2...

multipart.py

Source:multipart.py

...20    """Performs a multi-part upload to OCI object storage"""21    config = oci.config.from_file(config_path, config_profile)22    client = oci.object_storage.ObjectStorageClient(config)23    storage_namespace = client.get_namespace().data24    upload_id = create_multipart_upload(storage_namespace, bucket_name, object_name, client)25    part_number = 126    part_details_list = []27    for part in part_list:28        part_details = upload_part(storage_namespace, bucket_name, object_name, upload_id, part_number, part, client)29        part_details_list.append(part_details)30        part_number += 131    commit_multipart_upload(storage_namespace, bucket_name, object_name, upload_id, part_details_list, client)32def create_multipart_upload(storage_namespace, bucket_name, object_name, client):33    kwargs = { "object": object_name }34    details = oci.object_storage.models.CreateMultipartUploadDetails(**kwargs)35    upload_id = client.create_multipart_upload(storage_namespace, bucket_name, details).data.upload_id36    print('Upload ID: '+upload_id)37    return upload_id38def upload_part(storage_namespace, bucket_name, object_name, upload_id, part_number, part, client):39    print('Part File: '+part)40    kwargs = {}41    with open(part, 'rb') as part_stream:42        rsp = client.upload_part(storage_namespace, bucket_name, object_name, upload_id, part_number, part_stream)43        kwargs = { "part_num": part_number, "etag": rsp.headers['ETag'] }44        print('Part ETag: '+rsp.headers['ETag'])45    return oci.object_storage.models.CommitMultipartUploadPartDetails(**kwargs)46def commit_multipart_upload(storage_namespace, bucket_name, object_name, upload_id, part_details_list, client):47    kwargs = { "parts_to_commit": part_details_list }48    details = oci.object_storage.models.CommitMultipartUploadDetails(**kwargs)49    client.commit_multipart_upload(storage_namespace, bucket_name, object_name, upload_id, details)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.