Skip to content

Instantly share code, notes, and snippets.

@clayg
Created December 12, 2017 21:19
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save clayg/a883d21723bda1a947109be2d33e2be2 to your computer and use it in GitHub Desktop.
Save clayg/a883d21723bda1a947109be2d33e2be2 to your computer and use it in GitHub Desktop.
versioned writes functional tests
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