Merge pull request #1177 from lmlg/ceph-standalone

Make Ceph tests self-contained.
This commit is contained in:
Alex Kavanagh
2024-01-08 13:55:47 +00:00
committed by GitHub
2 changed files with 47 additions and 5 deletions
@@ -17,6 +17,7 @@ import json
import logging
import re
import time
import unittest
import cinderclient.exceptions as cinder_exceptions
@@ -154,7 +155,24 @@ def create_cinder_volume(cinder, name='zaza', image_id=None, type_id=None):
return create_volume(cinder, volume_params)
class CephRBDMirrorBase(test_utils.OpenStackBaseTest):
def setup_rbd_mirror():
"""Set up an RBD pool in case Cinder isn't present."""
zaza.model.run_action_on_leader(
'ceph-mon',
'create-pool',
action_params={
'name': 'zaza-boot',
'app-name': 'rbd',
}
)
zaza.model.run_action_on_leader(
'ceph-rbd-mirror',
'refresh-pools',
action_params={}
)
class CephRBDMirrorBase(test_utils.BaseCharmTest):
"""Base class for ``ceph-rbd-mirror`` tests."""
@classmethod
@@ -167,6 +185,20 @@ class CephRBDMirrorBase(test_utils.OpenStackBaseTest):
cls.site_a_model = cls.site_b_model = zaza.model.get_juju_model()
cls.site_b_app_suffix = '-b'
def test_if_cinder_present(self):
"""Test if the cinder-ceph application is present."""
try:
zaza.model.get_application(self.cinder_ceph_app_name)
return True
except KeyError:
return False
def skip_test_if_cinder_not_present(self, caller):
"""Skip a test if Cinder isn't present."""
if not self.test_if_cinder_present():
raise unittest.SkipTest('Skipping %s due to lack of Cinder'
% caller)
def run_status_action(self, application_name=None, model_name=None,
pools=[]):
"""Run status action, decode and return response."""
@@ -215,7 +247,9 @@ class CephRBDMirrorBase(test_utils.OpenStackBaseTest):
:rtype: Tuple[List[str], List[str]]
"""
site_a_pools, site_b_pools = self.get_pools()
if get_cinder_rbd_mirroring_mode(self.cinder_ceph_app_name) == 'image':
if (self.test_if_cinder_present() and
get_cinder_rbd_mirroring_mode(self.cinder_ceph_app_name) ==
'image'):
site_a_pools.remove(self.cinder_ceph_app_name)
site_b_pools.remove(self.cinder_ceph_app_name)
@@ -393,6 +427,7 @@ class CephRBDMirrorTest(CephRBDMirrorBase):
site B and subsequently comparing the contents we get a full end to end
test.
"""
self.skip_test_if_cinder_not_present('test_cinder_volume_mirrored')
volume = self.setup_test_cinder_volume()
site_a_hash = zaza.openstack.utilities.ceph.get_rbd_hash(
zaza.model.get_lead_unit_name('ceph-mon',
@@ -534,6 +569,7 @@ class CephRBDMirrorControlledFailoverTest(CephRBDMirrorBase):
This test only makes sense if Cinder RBD mirroring mode is 'image'.
It will return early, if this is not the case.
"""
self.skip_test_if_cinder_not_present('test_100_cinder_failover')
cinder_rbd_mirroring_mode = get_cinder_rbd_mirroring_mode(
self.cinder_ceph_app_name)
if cinder_rbd_mirroring_mode != 'image':
@@ -583,6 +619,7 @@ class CephRBDMirrorControlledFailoverTest(CephRBDMirrorBase):
The test needs to be executed when the Cinder volume host is already
failed-over with the test volume on it.
"""
self.skip_test_if_cinder_not_present('test_101_cinder_failback')
cinder_rbd_mirroring_mode = get_cinder_rbd_mirroring_mode(
self.cinder_ceph_app_name)
if cinder_rbd_mirroring_mode != 'image':
@@ -766,9 +803,9 @@ class CephRBDMirrorDisasterFailoverTest(CephRBDMirrorBase):
})
self.assertEqual(int(result.results['Code']), 0)
# The site-b 'promote' Juju action is expected to fail, because the
# primary site is down.
self.assertEqual(result.status, 'failed')
# The action may not show up as 'failed' if there are no pools that
# needed to be promoted.
# self.assertEqual(result.status, 'failed')
# Retry to promote site-b using the 'force' Juju action parameter.
result = zaza.model.run_action_on_leader(
@@ -792,6 +829,7 @@ class CephRBDMirrorDisasterFailoverTest(CephRBDMirrorBase):
This assumes that the primary site is already killed.
"""
self.skip_test_if_cinder_not_present('test_200_forced_cinder_failover')
cinder_rbd_mirroring_mode = get_cinder_rbd_mirroring_mode(
self.cinder_ceph_app_name)
if cinder_rbd_mirroring_mode != 'image':
+4
View File
@@ -1221,6 +1221,10 @@ class CephProxyTest(unittest.TestCase):
def test_cinder_ceph_restrict_pool_setup(self):
"""Make sure cinder-ceph restrict pool was created successfully."""
try:
zaza_model.get_application('cinder-ceph')
except KeyError:
raise unittest.SkipTest("Skipping OpenStack dependent test")
logging.info('Wait for idle/ready status...')
zaza_model.wait_for_application_states(
states=self.target_deploy_status)