Add Ceph test CheckPoolTypes

Add a ceph test to check that the type of pools requested by
clients matches the pools that were created.

zaza/openstack/charm_tests/ceph/tests.py

@@ -794,6 +794,50 @@ class CephPrometheusTest(unittest.TestCase):
             '3', _get_mon_count_from_prometheus(unit.public_address))
 
 
+class CephPoolConfig(Exception):
+    """Custom Exception for bad Ceph pool config."""
+
+    pass
+
+
+class CheckPoolTypes(unittest.TestCase):
+    """Test the ceph pools created for clients are of the expected type."""
+
+    def test_check_pool_types(self):
+        """Check type of pools created for clients."""
+        app_pools = [
+            ('glance', 'glance'),
+            ('nova-compute', 'nova'),
+            ('cinder-ceph', 'cinder-ceph')]
+        runtime_pool_details = zaza_ceph.get_ceph_pool_details()
+        for app, pool_name in app_pools:
+            juju_pool_config = zaza_model.get_application_config(app).get(
+                'pool-type')
+            if juju_pool_config:
+                expected_pool_type = juju_pool_config['value']
+            else:
+                # If the pool-type option is absent assume the default of
+                # replicated.
+                expected_pool_type = zaza_ceph.REPLICATED_POOL_TYPE
+            for pool_config in runtime_pool_details:
+                if pool_config['pool_name'] == pool_name:
+                    logging.info('Checking {} is {}'.format(
+                        pool_name,
+                        expected_pool_type))
+                    expected_pool_code = -1
+                    if expected_pool_type == zaza_ceph.REPLICATED_POOL_TYPE:
+                        expected_pool_code = zaza_ceph.REPLICATED_POOL_CODE
+                    elif expected_pool_type == zaza_ceph.ERASURE_POOL_TYPE:
+                        expected_pool_code = zaza_ceph.ERASURE_POOL_CODE
+                    self.assertEqual(
+                        pool_config['type'],
+                        expected_pool_code)
+                    break
+            else:
+                raise CephPoolConfig(
+                    "Failed to find config for {}".format(pool_name))
+
+
 # NOTE: We might query before prometheus has fetch data
 @tenacity.retry(wait=tenacity.wait_exponential(multiplier=1,
                                                min=5, max=10),

zaza/openstack/utilities/ceph.py

@@ -5,6 +5,11 @@ import logging
 import zaza.openstack.utilities.openstack as openstack_utils
 import zaza.model as zaza_model
 
+REPLICATED_POOL_TYPE = 'replicated'
+ERASURE_POOL_TYPE = 'erasure-coded'
+REPLICATED_POOL_CODE = 1
+ERASURE_POOL_CODE = 3
+
 
 def get_expected_pools(radosgw=False):
     """Get expected ceph pools.
@@ -97,6 +102,39 @@ def get_ceph_pools(unit_name, model_name=None):
     return pools
 
 
+def get_ceph_pool_details(query_leader=True, unit_name=None, model_name=None):
+    """Get ceph pool details.
+
+    Return a list of ceph pools details dicts.
+
+    :param query_leader: Whether to query the leader for pool details.
+    :type query_leader: bool
+    :param unit_name: Name of unit to get the pools on if query_leader is False
+    :type unit_name: string
+    :param model_name: Name of model to operate in
+    :type model_name: str
+    :returns: Dict of ceph pools
+    :rtype: List[Dict,]
+    :raise: zaza_model.CommandRunFailed
+    """
+    cmd = 'sudo ceph osd pool ls detail -f json'
+    if query_leader and unit_name:
+        raise ValueError("Cannot set query_leader and unit_name")
+    if query_leader:
+        result = zaza_model.run_on_leader(
+            'ceph-mon',
+            cmd,
+            model_name=model_name)
+    else:
+        result = zaza_model.run_on_unit(
+            unit_name,
+            cmd,
+            model_name=model_name)
+    if int(result.get('Code')) != 0:
+        raise zaza_model.CommandRunFailed(cmd, result)
+    return json.loads(result.get('Stdout'))
+
+
 def get_ceph_df(unit_name, model_name=None):
     """Return dict of ceph df json output, including ceph pool state.