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

Fixed copy-object from unversioned bucket to versioned bucket

Browse Source 
The response of the copy-object operation was missing VersionId property
when source bucket is not versioned.
This commit is contained in:
Yaroslav Admin 2019-04-02 15:30:01 +02:00
parent 66dafc6679
commit b85d21b8fe
2 changed files with 27 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
moto/s3/models.py
View File

@ -87,10 +87,13 @@ class FakeKey(BaseModel):
new_value = new_value.encode(DEFAULT_TEXT_ENCODING)
self._value_buffer.write(new_value)
def copy(self, new_name=None):
def copy(self, new_name=None, new_is_versioned=None):
r = copy.deepcopy(self)
if new_name is not None:
r.name = new_name
if new_is_versioned is not None:
r._is_versioned = new_is_versioned
r.refresh_version()
return r
def set_metadata(self, metadata, replace=False):
@ -973,17 +976,15 @@ class S3Backend(BaseBackend):
dest_bucket = self.get_bucket(dest_bucket_name)
key = self.get_key(src_bucket_name, src_key_name,
version_id=src_version_id)
if dest_key_name != src_key_name:
key = key.copy(dest_key_name)
dest_bucket.keys[dest_key_name] = key
# By this point, the destination key must exist, or KeyError
if dest_bucket.is_versioned:
dest_bucket.keys[dest_key_name].refresh_version()
new_key = key.copy(dest_key_name, dest_bucket.is_versioned)
if storage is not None:
key.set_storage_class(storage)
new_key.set_storage_class(storage)
if acl is not None:
key.set_acl(acl)
new_key.set_acl(acl)
dest_bucket.keys[dest_key_name] = new_key
def set_bucket_acl(self, bucket_name, acl):
bucket = self.get_bucket(bucket_name)

17
tests/test_s3/test_s3.py
View File

@ -1530,6 +1530,23 @@ def test_boto3_copy_object_with_versioning():
obj2_version_new.should_not.equal(obj2_version)
@mock_s3
def test_boto3_copy_object_from_unversioned_to_versioned_bucket():
client = boto3.client('s3', region_name='us-east-1')
client.create_bucket(Bucket='src', CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'})
client.create_bucket(Bucket='dest', CreateBucketConfiguration={'LocationConstraint': 'eu-west-1'})
client.put_bucket_versioning(Bucket='dest', VersioningConfiguration={'Status': 'Enabled'})
client.put_object(Bucket='src', Key='test', Body=b'content')
obj2_version_new = client.copy_object(CopySource={'Bucket': 'src', 'Key': 'test'}, Bucket='dest', Key='test') \
.get('VersionId')
# VersionId should be present in the response
obj2_version_new.should_not.equal(None)
@mock_s3
def test_boto3_deleted_versionings_list():
client = boto3.client('s3', region_name='us-east-1')