diff --git a/moto/s3/responses.py b/moto/s3/responses.py index e653b0b6862f..e875d7594746 100644 --- a/moto/s3/responses.py +++ b/moto/s3/responses.py @@ -2533,9 +2533,12 @@ def _key_response_post( multipart_id = query["uploadId"][0] - if existing is not None and existing.multipart: - # Based on testing against AWS, operation seems idempotent - # Scenario where both method-calls have a different body hasn't been tested yet + if ( + existing is not None + and existing.multipart + and existing.multipart.id == multipart_id + ): + # Operation is idempotent key: Optional[FakeKey] = existing else: key = self.backend.complete_multipart_upload( diff --git a/tests/test_s3/test_s3_multipart.py b/tests/test_s3/test_s3_multipart.py index 9859f324cedc..f1e4e3fef44b 100644 --- a/tests/test_s3/test_s3_multipart.py +++ b/tests/test_s3/test_s3_multipart.py @@ -1142,3 +1142,55 @@ def test_generate_presigned_url_for_multipart_upload(): resp = requests.put(url, data=data) assert resp.status_code == 200 + + +@s3_aws_verified +@pytest.mark.aws_verified() +def test_multipart_upload_overwrites(bucket_name=None): + s3_client = boto3.client("s3", "us-east-1") + key = "mykey.txt" + + upload_id = s3_client.create_multipart_upload(Bucket=bucket_name, Key=key)[ + "UploadId" + ] + + part_data = b"First part data" + response = s3_client.upload_part( + Bucket=bucket_name, + Key=key, + PartNumber=1, + UploadId=upload_id, + Body=part_data, + ) + + s3_client.complete_multipart_upload( + Bucket=bucket_name, + Key=key, + UploadId=upload_id, + MultipartUpload={"Parts": [{"PartNumber": 1, "ETag": response["ETag"]}]}, + ) + + upload_id = s3_client.create_multipart_upload(Bucket=bucket_name, Key=key)[ + "UploadId" + ] + + new_data = b"New data that should overwrite" + response = s3_client.upload_part( + Bucket=bucket_name, + Key=key, + PartNumber=1, + UploadId=upload_id, + Body=new_data, + ) + + s3_client.complete_multipart_upload( + Bucket=bucket_name, + Key=key, + UploadId=upload_id, + MultipartUpload={"Parts": [{"PartNumber": 1, "ETag": response["ETag"]}]}, + ) + + result = s3_client.get_object(Bucket=bucket_name, Key=key) + content = result["Body"].read() + + assert content == new_data