diff --git a/test/functional/test_versioned_writes.py b/test/functional/test_versioned_writes.py index 2dd2c0ba81..e002a39b00 100644 --- a/test/functional/test_versioned_writes.py +++ b/test/functional/test_versioned_writes.py @@ -209,9 +209,12 @@ class TestObjectVersioning(Base): try: # only delete files and not containers # as they were configured in self.env + # get rid of any versions so they aren't restored self.env.versions_container.delete_files() + # get rid of originals self.env.container.delete_files() - # in history mode, cleanup after deleting 'container' files + # in history mode, deleted originals got copied to versions, so + # clear that again self.env.versions_container.delete_files() except ResponseError: pass @@ -630,6 +633,69 @@ class TestObjectVersioning(Base): self.assertEqual('%s/%s' % (self.env.container.name, target.name), sym_info['x_symlink_target']) + def _setup_symlink(self): + target = self.env.container.file('target-object') + target.write('target object data') + symlink = self.env.container.file('symlink') + symlink.write('', hdrs={ + 'Content-Type': 'application/symlink', + 'X-Symlink-Target': '%s/%s' % ( + self.env.container.name, target.name)}) + return symlink, target + + def _assert_symlink(self, symlink, target): + self.assertEqual('target object data', symlink.read()) + self.assertEqual(target.info(), symlink.info()) + self.assertEqual('application/symlink', + symlink.info(parms={ + 'symlink': 'get'})['content_type']) + + def _check_copy_destination_restore_symlink(self): + # assertions common to x-versions-location and x-history-location modes + symlink, target = self._setup_symlink() + symlink.write('this is not a symlink') + # the symlink is versioned + version_container_files = self.env.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.env.versions_container.file( + versioned_obj_info['name']) + # the symlink is still a symlink + self._assert_symlink(versioned_obj, target) + # test manual restore (this creates a new backup of the overwrite) + versioned_obj.copy(self.env.container.name, symlink.name, + parms={'symlink': 'get'}) + self._assert_symlink(symlink, target) + # symlink overwritten by write then copy -> 2 versions + self.assertEqual(2, self.env.versions_container.info()['object_count']) + return symlink, target + + def test_copy_destination_restore_symlink(self): + symlink, target = self._check_copy_destination_restore_symlink() + # and versioned writes restore + symlink.delete() + self.assertEqual(1, self.env.versions_container.info()['object_count']) + self.assertEqual('this is not a symlink', symlink.read()) + symlink.delete() + self.assertEqual(0, self.env.versions_container.info()['object_count']) + self._assert_symlink(symlink, target) + + def test_put_x_copy_from_restore_symlink(self): + symlink, target = self._setup_symlink() + symlink.write('this is not a symlink') + version_container_files = self.env.versions_container.files() + self.assertEqual(1, len(version_container_files)) + versioned_obj = self.env.versions_container.file( + version_container_files[0]) + symlink.write(parms={'symlink': 'get'}, cfg={ + 'no_content_type': True}, hdrs={ + 'X-Copy-From': '%s/%s' % ( + self.env.versions_container, versioned_obj.name)}) + self._assert_symlink(symlink, target) + class TestObjectVersioningUTF8(Base2, TestObjectVersioning): @@ -751,6 +817,15 @@ class TestObjectVersioningHistoryMode(TestObjectVersioning): symlink.read() self.assertEqual(404, cm.exception.status) + def test_copy_destination_restore_symlink(self): + symlink, target = self._check_copy_destination_restore_symlink() + symlink.delete() + with self.assertRaises(ResponseError) as cm: + symlink.read() + self.assertEqual(404, cm.exception.status) + # 2 versions plus delete marker and deleted version + self.assertEqual(4, self.env.versions_container.info()['object_count']) + class TestSloWithVersioning(unittest2.TestCase):