diff --git a/moto/s3/models.py b/moto/s3/models.py index 9e023f68d..d4506678b 100644 --- a/moto/s3/models.py +++ b/moto/s3/models.py @@ -474,12 +474,12 @@ class S3Backend(BaseBackend): bucket = self.get_bucket(bucket_name) return bucket.keys.pop(key_name) - def copy_key(self, src_bucket_name, src_key_name, dest_bucket_name, dest_key_name, storage=None, acl=None): + def copy_key(self, src_bucket_name, src_key_name, dest_bucket_name, + dest_key_name, storage=None, acl=None, src_version_id=None): src_key_name = clean_key_name(src_key_name) dest_key_name = clean_key_name(dest_key_name) - src_bucket = self.get_bucket(src_bucket_name) dest_bucket = self.get_bucket(dest_bucket_name) - key = src_bucket.keys[src_key_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 diff --git a/moto/s3/responses.py b/moto/s3/responses.py index f27606c90..0339d6853 100644 --- a/moto/s3/responses.py +++ b/moto/s3/responses.py @@ -435,9 +435,11 @@ class ResponseObject(_TemplateEnvironmentMixin): if 'x-amz-copy-source' in request.headers: # Copy key - src_bucket, src_key = request.headers.get("x-amz-copy-source").split("/", 1) + src_key_parsed = urlparse(request.headers.get("x-amz-copy-source")) + src_bucket, src_key = src_key_parsed.path.split("/", 1) + src_version_id = parse_qs(src_key_parsed.query).get('versionId', [None])[0] self.backend.copy_key(src_bucket, src_key, bucket_name, key_name, - storage=storage_class, acl=acl) + storage=storage_class, acl=acl, src_version_id=src_version_id) mdirective = request.headers.get('x-amz-metadata-directive') if mdirective is not None and mdirective == 'REPLACE': new_key = self.backend.get_key(bucket_name, key_name) diff --git a/tests/test_s3/test_s3.py b/tests/test_s3/test_s3.py index e2de0c473..88eb3f985 100644 --- a/tests/test_s3/test_s3.py +++ b/tests/test_s3/test_s3.py @@ -332,6 +332,22 @@ def test_copy_key(): bucket.get_key("new-key").get_contents_as_string().should.equal(b"some value") +@mock_s3 +def test_copy_key_with_version(): + conn = boto.connect_s3('the_key', 'the_secret') + bucket = conn.create_bucket("foobar") + bucket.configure_versioning(versioning=True) + key = Key(bucket) + key.key = "the-key" + key.set_contents_from_string("some value") + key.set_contents_from_string("another value") + + bucket.copy_key('new-key', 'foobar', 'the-key', src_version_id='0') + + bucket.get_key("the-key").get_contents_as_string().should.equal(b"another value") + bucket.get_key("new-key").get_contents_as_string().should.equal(b"some value") + + @mock_s3 def test_set_metadata(): conn = boto.connect_s3('the_key', 'the_secret')