openstack/heat
Code Issues Proposed changes

Merge "Handle missing volume in handle_snapshot_delete"

Browse Source
This commit is contained in:
Zuul
2025-08-22 17:03:49 +00:00
committed by Gerrit Code Review
parent 8efc5c6a76 1d71efa038
commit 6fb9d3e967
2 changed files with 99 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
heat/engine/resources/volume_base.py
View File

@@ -12,6 +12,7 @@
# under the License.
from oslo_config import cfg
from oslo_log import log as logging
from heat.common import exception
from heat.common.i18n import _
@@ -19,6 +20,8 @@ from heat.engine.clients import progress
from heat.engine import resource
from heat.engine import rsrc_defn
LOG = logging.getLogger(__name__)
class BaseVolume(resource.Resource):
"""Base Volume Manager."""
@@ -95,8 +98,13 @@ class BaseVolume(resource.Resource):
return prg
def _create_backup(self):
backup = self.client().backups.create(self.resource_id)
return backup.id
with self.client_plugin().ignore_not_found:
backup = self.client().backups.create(self.resource_id)
return backup.id
LOG.debug(
"Skip creating a backup for volume %s - NotFound",
self.resource_id)
return None
def _check_create_backup_complete(self, prg):
backup = self.client().backups.get(prg.backup_id)
@@ -133,6 +141,8 @@ class BaseVolume(resource.Resource):
if not prg.backup['called']:
prg.backup_id = self._create_backup()
prg.backup['called'] = True
if prg.backup_id is None:
prg.backup['complete'] = True
return False
if not prg.backup['complete']:

87
heat/tests/openstack/cinder/test_volume.py
View File

@@ -25,6 +25,7 @@ from heat.common import template_format
from heat.engine.clients.os import cinder
from heat.engine.clients.os import glance
from heat.engine.clients.os import nova
from heat.engine.clients import progress
from heat.engine.resources.openstack.cinder import volume as c_vol
from heat.engine.resources import scheduler_hints as sh
from heat.engine import rsrc_defn
@@ -1236,6 +1237,92 @@ class CinderVolumeTest(vt_base.VolumeTestCase):
stack)
self.assertIsNone(rsrc.handle_delete_snapshot(mock_vs))
def test_check_delete_complete(self):
fb_creating = vt_base.FakeBackup('creating')
fb_available = vt_base.FakeBackup("available")
self.cinder_fc.backups.create.return_value = fb_creating
self.cinder_fc.backups.get.side_effect = [fb_creating, fb_available]
self.cinder_fc.volumes.get.side_effect = [
vt_base.FakeVolume('available'),
vt_base.FakeVolume('deleting'),
cinder_exp.NotFound("Nope")
]
# will need backup first
fake_prg = progress.VolumeDeleteProgress()
self.stack_name = 'test_volume_check_delete'
t = template_format.parse(single_cinder_volume_template)
stack = utils.parse_stack(t, stack_name=self.stack_name)
rsrc = c_vol.CinderVolume(
'volume',
stack.t.resource_definitions(stack)['volume'],
stack)
# first backup was called
self.assertFalse(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertFalse(fake_prg.backup["complete"])
self.assertFalse(fake_prg.delete["called"])
self.assertFalse(fake_prg.delete["complete"])
self.assertEqual(fake_prg.backup_id, fb_creating.id)
self.cinder_fc.backups.create.assert_called_once_with(rsrc.resource_id)
# polling for backup status
self.assertFalse(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertFalse(fake_prg.backup["complete"])
self.assertFalse(fake_prg.delete["called"])
self.assertFalse(fake_prg.delete["complete"])
self.cinder_fc.backups.get.assert_called_once_with(fb_creating.id)
self.assertFalse(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertTrue(fake_prg.backup["complete"])
self.assertFalse(fake_prg.delete["called"])
self.assertFalse(fake_prg.delete["complete"])
self.cinder_fc.backups.get.assert_called_with(fb_creating.id)
self.assertEqual(2, self.cinder_fc.backups.get.call_count)
# volume delete called, includes one volume get too
self.assertFalse(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertTrue(fake_prg.backup["complete"])
self.assertTrue(fake_prg.delete["called"])
self.assertFalse(fake_prg.delete["complete"])
self.cinder_fc.volumes.delete.assert_called_once_with(rsrc.resource_id)
self.cinder_fc.volumes.get.assert_called_with(rsrc.resource_id)
self.assertEqual(1, self.cinder_fc.volumes.get.call_count)
# polling for volume deleted
self.assertFalse(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertTrue(fake_prg.backup["complete"])
self.assertTrue(fake_prg.delete["called"])
self.assertFalse(fake_prg.delete["complete"])
self.cinder_fc.volumes.get.assert_called_with(rsrc.resource_id)
self.assertEqual(2, self.cinder_fc.volumes.get.call_count)
self.assertTrue(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertTrue(fake_prg.backup["complete"])
self.assertTrue(fake_prg.delete["called"])
self.assertTrue(fake_prg.delete["complete"])
self.cinder_fc.volumes.get.assert_called_with(rsrc.resource_id)
self.assertEqual(3, self.cinder_fc.volumes.get.call_count)
def test_check_delete_complete_snapshot_missing_volume(self):
self.cinder_fc.backups.create.side_effect = cinder_exp.NotFound("Nope")
fake_prg = progress.VolumeDeleteProgress()
self.stack_name = 'test_check_delete_complete_snapshot_missing_volume'
t = template_format.parse(single_cinder_volume_template)
stack = utils.parse_stack(t, stack_name=self.stack_name)
rsrc = c_vol.CinderVolume(
'volume',
stack.t.resource_definitions(stack)['volume'],
stack)
self.assertFalse(rsrc.check_delete_complete(fake_prg))
self.assertTrue(fake_prg.backup["called"])
self.assertTrue(fake_prg.backup["complete"])
self.assertIsNone(fake_prg.backup_id)
self.cinder_fc.backups.create.assert_called_once_with(rsrc.resource_id)
def test_vaildate_deletion_policy(self):
cfg.CONF.set_override('backups_enabled', False, group='volumes')
self.stack_name = 'test_volume_validate_deletion_policy'