Merge pull request boto#2803 from sblosser/patch-1

Pass version_id in copy if key is versioned. Fixes boto#2803.
stanhu · Dec 18, 2014 · 66b3604 · 66b3604
2 parents 90a5d44 + 230cf17
commit 66b3604
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 1 deletion.
diff --git a/boto/s3/key.py b/boto/s3/key.py
@@ -496,7 +496,8 @@ def copy(self, dst_bucket, dst_key, metadata=None,
                                    self.name, metadata,
                                    storage_class=storage_class,
                                    preserve_acl=preserve_acl,
-                                   encrypt_key=encrypt_key)
+                                   encrypt_key=encrypt_key,
+                                   src_version_id=self.version_id)
 
     def startElement(self, name, attrs, connection):
         if name == 'Owner':

diff --git a/tests/integration/s3/test_key.py b/tests/integration/s3/test_key.py
@@ -487,3 +487,48 @@ def test_put_get_with_non_string_headers_key(self):
         from_s3_key = self.bucket.get_key('foobar', headers=headers)
         self.assertEqual(from_s3_key.get_contents_as_string().decode('utf-8'),
                          body)
+
+
+class S3KeyVersionCopyTest(unittest.TestCase):
+    def setUp(self):
+        self.conn = S3Connection()
+        self.bucket_name = 'boto-key-version-copy-%d' % int(time.time())
+        self.bucket = self.conn.create_bucket(self.bucket_name)
+        self.bucket.configure_versioning(True)
+
+    def tearDown(self):
+        for key in self.bucket.list_versions():
+            key.delete()
+        self.bucket.delete()
+
+    def test_key_overwrite_and_copy(self):
+        first_content = "abcdefghijklm"
+        second_content = "nopqrstuvwxyz"
+        k = Key(self.bucket, 'testkey')
+        k.set_contents_from_string(first_content)
+        # Wait for S3's eventual consistency (may not be necessary)
+        while self.bucket.get_key('testkey') is None:
+            time.sleep(5)
+        # Get the first version_id
+        first_key = self.bucket.get_key('testkey')
+        first_version_id = first_key.version_id
+        # Overwrite the key
+        k = Key(self.bucket, 'testkey')
+        k.set_contents_from_string(second_content)
+        # Wait for eventual consistency
+        while True:
+            second_key = self.bucket.get_key('testkey')
+            if second_key is None or second_key.version_id == first_version_id:
+                time.sleep(5)
+            else:
+                break
+        # Copy first key (no longer the current version) to a new key
+        source_key = self.bucket.get_key('testkey',
+                                          version_id=first_version_id)
+        source_key.copy(self.bucket, 'copiedkey')
+        while self.bucket.get_key('copiedkey') is None:
+            time.sleep(5)
+        copied_key = self.bucket.get_key('copiedkey')
+        copied_key_contents = copied_key.get_contents_as_string()
+        self.assertEqual(first_content, copied_key_contents)
+