From a4b71877e4f0b13d2359e33574d241354aae692a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arpad=20M=C3=BCller?= Date: Sun, 4 Feb 2024 22:56:27 +0100 Subject: [PATCH] S3: Allow in place copies when bucket versioning is enabled (#7303) --- moto/s3/models.py | 5 +-- tests/test_s3/test_s3_copyobject.py | 52 +++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 2 deletions(-) diff --git a/moto/s3/models.py b/moto/s3/models.py index 291459bcd..4365f1e19 100644 --- a/moto/s3/models.py +++ b/moto/s3/models.py @@ -108,7 +108,7 @@ class FakeKey(BaseModel, ManagedState): storage: Optional[str] = "STANDARD", etag: Optional[str] = None, is_versioned: bool = False, - version_id: str = "null", + version_id: Optional[str] = None, max_buffer_size: Optional[int] = None, multipart: Optional["FakeMultipart"] = None, bucket_name: Optional[str] = None, @@ -170,7 +170,7 @@ class FakeKey(BaseModel, ManagedState): @property def version_id(self) -> str: - return self._version_id + return self._version_id or "null" @property def value(self) -> bytes: @@ -2674,6 +2674,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider): mdirective == "REPLACE", website_redirect_location, bucket.encryption, # S3 will allow copy in place if the bucket has encryption configured + src_key._version_id and bucket.is_versioned, ) ): raise CopyObjectMustChangeSomething diff --git a/tests/test_s3/test_s3_copyobject.py b/tests/test_s3/test_s3_copyobject.py index 527cda3dd..c734d22c2 100644 --- a/tests/test_s3/test_s3_copyobject.py +++ b/tests/test_s3/test_s3_copyobject.py @@ -783,6 +783,58 @@ def test_copy_object_in_place_with_bucket_encryption(): assert response["ServerSideEncryption"] == "AES256" +@mock_aws +def test_copy_object_in_place_with_versioning(): + # If a bucket has versioning enabled, it will allow copy in place + client = boto3.client("s3", region_name=DEFAULT_REGION_NAME) + bucket_name = "testbucket" + client.create_bucket(Bucket=bucket_name) + key = "source-key" + + response = client.put_object( + Body=b"", + Bucket=bucket_name, + Key=key, + ) + + response = client.put_bucket_versioning( + Bucket=bucket_name, + VersioningConfiguration={ + "MFADelete": "Disabled", + "Status": "Enabled", + }, + ) + assert response["ResponseMetadata"]["HTTPStatusCode"] == 200 + + response = client.put_object( + Body=b"", + Bucket=bucket_name, + Key=key, + ) + version_id = response["ResponseMetadata"]["HTTPHeaders"]["x-amz-version-id"] + assert version_id and version_id != "null" + + response = client.copy_object( + Bucket=bucket_name, + CopySource={"Bucket": bucket_name, "Key": key, "VersionId": version_id}, + Key=key, + ) + assert response["ResponseMetadata"]["HTTPStatusCode"] == 200 + + response = client.copy_object( + Bucket=bucket_name, + CopySource={"Bucket": bucket_name, "Key": key, "VersionId": "null"}, + Key=key, + ) + assert response["ResponseMetadata"]["HTTPStatusCode"] == 200 + + response = client.list_object_versions( + Bucket=bucket_name, + Prefix=key, + ) + assert len(response["Versions"]) == 4 + + @mock_aws @pytest.mark.parametrize( "algorithm",