Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix copying S3 objects, when using MetadataDirective='REPLACE' (#4883)

Browse Source
This commit is contained in:
Amit Shah 2022-02-24 14:21:27 +00:00 committed by GitHub
parent b6f4584737
commit e84cc20abe
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 29 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
moto/s3/models.py
View File

@ -2016,6 +2016,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
encryption=None,
kms_key_id=None,
bucket_key_enabled=False,
mdirective=None,
):
if (
src_key.name == dest_key_name
@ -2025,6 +2026,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
and encryption == src_key.encryption
and kms_key_id == src_key.kms_key_id
and bucket_key_enabled == (src_key.bucket_key_enabled or False)
and mdirective != "REPLACE"
):
raise CopyObjectMustChangeSomething

4
moto/s3/responses.py
View File

@ -1468,6 +1468,8 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
== "true"
)
mdirective = request.headers.get("x-amz-metadata-directive")
self.backend.copy_object(
key,
bucket_name,
@ -1477,12 +1479,12 @@ class ResponseObject(_TemplateEnvironmentMixin, ActionAuthenticatorMixin):
kms_key_id=kms_key_id,
encryption=encryption,
bucket_key_enabled=bucket_key_enabled,
mdirective=mdirective,
)
else:
raise MissingKey(key=src_key)
new_key = self.backend.get_object(bucket_name, key_name)
mdirective = request.headers.get("x-amz-metadata-directive")
if mdirective is not None and mdirective == "REPLACE":
metadata = metadata_from_headers(request.headers)
new_key.set_metadata(metadata, replace=True)

24
tests/test_s3/test_s3.py
View File

@ -704,6 +704,30 @@ def test_copy_key_without_changes_should_error_boto3():
)
@mock_s3
def test_copy_key_without_changes_should_not_error_boto3():
client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
bucket_name = "my_bucket"
s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
key_name = "my_key"
key = s3.Object(bucket_name, key_name)
s3.create_bucket(Bucket=bucket_name)
key.put(Body=b"some value")
client.copy_object(
Bucket=bucket_name,
CopySource="{}/{}".format(bucket_name, key_name),
Key=key_name,
Metadata={"some-key": "some-value"},
MetadataDirective="REPLACE",
)
new_object = client.get_object(Bucket=bucket_name, Key=key_name)
assert new_object["Metadata"] == {"some-key": "some-value"}
@freeze_time("2012-01-01 12:00:00")
@mock_s3
def test_last_modified_boto3():