Created
December 12, 2017 21:19
-
-
Save clayg/a883d21723bda1a947109be2d33e2be2 to your computer and use it in GitHub Desktop.
versioned writes functional tests
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
diff --git a/test/functional/test_versioned_writes.py b/test/functional/test_versioned_writes.py | |
index 9cd7b3b..4dc325f 100644 | |
--- a/test/functional/test_versioned_writes.py | |
+++ b/test/functional/test_versioned_writes.py | |
@@ -794,3 +794,87 @@ class TestSloWithVersioning(unittest2.TestCase): | |
# expect the original manifest file to be restored | |
self._assert_is_manifest(file_item, 'a') | |
self._assert_is_object(file_item, 'a') | |
+ | |
+ | |
+class TestSymlinkWithVersioning(unittest2.TestCase): | |
+ | |
+ def setUp(self): | |
+ if 'symlink' not in cluster_info: | |
+ raise SkipTest("Symlinks not enabled") | |
+ if tf.in_process: | |
+ tf.skip_if_no_xattrs() | |
+ self.conn = Connection(tf.config) | |
+ self.conn.authenticate() | |
+ self.account = Account( | |
+ self.conn, tf.config.get('account', tf.config['username'])) | |
+ self.account.delete_containers() | |
+ self.versions_container = self.account.container(Utils.create_name()) | |
+ self.container = self.account.container(Utils.create_name()) | |
+ if not self.container.create( | |
+ hdrs={'X-Versions-Location': self.versions_container.name}): | |
+ raise ResponseError(self.conn.response) | |
+ if 'versions' not in self.container.info(): | |
+ raise SkipTest("Object versioning not enabled") | |
+ if not self.versions_container.create(): | |
+ raise ResponseError(self.conn.response) | |
+ | |
+ self.target = self.container.file('target-object') | |
+ self.target.write('target object data') | |
+ self.symlink = self.container.file('symlink') | |
+ self.symlink.write('', hdrs={ | |
+ 'Content-Type': 'application/symlink', | |
+ 'x-symlink-target': '%s/%s' % ( | |
+ self.container.name, self.target.name)}) | |
+ | |
+ def test_copy_destination_restore_symlink(self): | |
+ self.symlink.write('this is not a symlink') | |
+ # the symlink is versioned | |
+ version_container_files = self.versions_container.files( | |
+ parms={'format': 'json'}) | |
+ self.assertEqual(1, len(version_container_files)) | |
+ versioned_obj_info = version_container_files[0] | |
+ self.assertEqual('application/symlink', | |
+ versioned_obj_info['content_type']) | |
+ versioned_obj = self.versions_container.file( | |
+ versioned_obj_info['name']) | |
+ # the symlink is still a symlink | |
+ self.assertEqual('target object data', versioned_obj.read()) | |
+ self.assertEqual(self.target.info(), versioned_obj.info()) | |
+ self.assertEqual('application/symlink', | |
+ versioned_obj.info(parms={ | |
+ 'symlink': 'get'})['content_type']) | |
+ # test manual restore (this creates a new backup of the overwrite) | |
+ versioned_obj.copy(self.container.name, self.symlink.name, | |
+ parms={'symlink': 'get'}) | |
+ self.assertEqual('target object data', self.symlink.read()) | |
+ self.assertEqual(self.target.info(), self.symlink.info()) | |
+ self.assertEqual('application/symlink', | |
+ self.symlink.info(parms={ | |
+ 'symlink': 'get'})['content_type']) | |
+ # and versioned writes restore | |
+ self.symlink.delete() | |
+ self.assertEqual(1, self.versions_container.info()['object_count']) | |
+ self.assertEqual('this is not a symlink', self.symlink.read()) | |
+ self.symlink.delete() | |
+ self.assertEqual(0, self.versions_container.info()['object_count']) | |
+ self.assertEqual('target object data', self.symlink.read()) | |
+ self.assertEqual(self.target.info(), self.symlink.info()) | |
+ self.assertEqual('application/symlink', | |
+ self.symlink.info(parms={ | |
+ 'symlink': 'get'})['content_type']) | |
+ | |
+ def test_put_x_copy_from_restore_symlink(self): | |
+ self.symlink.write('this is not a symlink') | |
+ version_container_files = self.versions_container.files() | |
+ self.assertEqual(1, len(version_container_files)) | |
+ versioned_obj = self.versions_container.file( | |
+ version_container_files[0]) | |
+ self.symlink.write(parms={'symlink': 'get'}, cfg={ | |
+ 'no_content_type': True}, hdrs={ | |
+ 'x-copy-from': '%s/%s' % ( | |
+ self.versions_container, versioned_obj.name)}) | |
+ self.assertEqual('target object data', self.symlink.read()) | |
+ self.assertEqual(self.target.info(), self.symlink.info()) | |
+ self.assertEqual('application/symlink', | |
+ self.symlink.info(parms={ | |
+ 'symlink': 'get'})['content_type']) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment