Merge "Make scenario snapshot tests work with ephemeral|swap"

This commit is contained in:
Zuul 2024-11-20 20:21:37 +00:00 committed by Gerrit Code Review
commit e7b104c23c
2 changed files with 27 additions and 17 deletions

View File

@ -923,6 +923,19 @@ class ScenarioTest(tempest.test.BaseTestCase):
if not isinstance(exc, lib_exc.SSHTimeout): if not isinstance(exc, lib_exc.SSHTimeout):
LOG.debug('Network information on a devstack host') LOG.debug('Network information on a devstack host')
def get_snapshot_id(self, bdms):
if isinstance(bdms, str):
bdms = json.loads(bdms)
snapshot_id = None
for bdm in bdms:
# Look for the block device mapping that actually has a
# snapshot. If the server has ephemeral or swap disk, their
# block device mappings will be present with snapshot_id = None
if 'snapshot_id' in bdm and bdm['snapshot_id'] is not None:
snapshot_id = bdm['snapshot_id']
break
return snapshot_id
def create_server_snapshot(self, server, name=None, **kwargs): def create_server_snapshot(self, server, name=None, **kwargs):
"""Creates server snapshot""" """Creates server snapshot"""
# Glance client # Glance client
@ -949,20 +962,19 @@ class ScenarioTest(tempest.test.BaseTestCase):
snapshot_image = _image_client.show_image(image_id) snapshot_image = _image_client.show_image(image_id)
image_props = snapshot_image image_props = snapshot_image
bdm = image_props.get('block_device_mapping') bdms = image_props.get('block_device_mapping')
if bdm: if bdms:
bdm = json.loads(bdm) snapshot_id = self.get_snapshot_id(bdms)
if bdm and 'snapshot_id' in bdm[0]: self.assertIsNotNone(snapshot_id)
snapshot_id = bdm[0]['snapshot_id']
self.addCleanup( self.addCleanup(
self.snapshots_client.wait_for_resource_deletion, self.snapshots_client.wait_for_resource_deletion,
snapshot_id) snapshot_id)
self.addCleanup(test_utils.call_and_ignore_notfound_exc, self.addCleanup(test_utils.call_and_ignore_notfound_exc,
self.snapshots_client.delete_snapshot, self.snapshots_client.delete_snapshot,
snapshot_id) snapshot_id)
waiters.wait_for_volume_resource_status(self.snapshots_client, waiters.wait_for_volume_resource_status(
snapshot_id, self.snapshots_client, snapshot_id, 'available')
'available')
image_name = snapshot_image['name'] image_name = snapshot_image['name']
self.assertEqual(name, image_name) self.assertEqual(name, image_name)
LOG.debug("Created snapshot image %s for server %s", LOG.debug("Created snapshot image %s for server %s",

View File

@ -11,7 +11,6 @@
# under the License. # under the License.
from oslo_log import log as logging from oslo_log import log as logging
from oslo_serialization import jsonutils as json
import testtools import testtools
from tempest.common import utils from tempest.common import utils
@ -245,8 +244,7 @@ class TestVolumeBootPattern(manager.EncryptionScenarioTest):
bdms = image.get('block_device_mapping') bdms = image.get('block_device_mapping')
if not bdms: if not bdms:
bdms = image['properties']['block_device_mapping'] bdms = image['properties']['block_device_mapping']
bdms = json.loads(bdms) snapshot_id = self.get_snapshot_id(bdms)
snapshot_id = bdms[0]['snapshot_id']
self._delete_snapshot(snapshot_id) self._delete_snapshot(snapshot_id)
# Now, delete the first server which will also delete the first # Now, delete the first server which will also delete the first