From 2fd6b31aff41e9513c0a937ca307e4e01112988c Mon Sep 17 00:00:00 2001
From: Bert Blommers <info@bertblommers.nl>
Date: Wed, 18 Dec 2024 18:58:01 -0100
Subject: [PATCH] S3: complete_multipart_upload() should be able to overwrite

---
 moto/s3/responses.py               |  9 ++++--
 tests/test_s3/test_s3_multipart.py | 52 ++++++++++++++++++++++++++++++
 2 files changed, 58 insertions(+), 3 deletions(-)

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