Merge pull request #1195 from lmlg/ceph-key-tests

Implement tests for ceph-mon's rotate-key action

zaza/openstack/charm_tests/ceph/tests.py

@@ -1748,3 +1748,67 @@ class CephMonJujuPersistent(test_utils.BaseCharmTest):
             )
             data = json.loads(result['Stdout'])
             assert data['loglevel'] == 2
+
+
+class CephMonKeyRotationTests(test_utils.BaseCharmTest):
+    """Tests for the rotate-key action."""
+
+    def _get_all_keys(self, unit, entity_filter):
+        cmd = 'sudo ceph auth ls'
+        result = zaza_model.run_on_unit(unit, cmd)
+        # Don't use json formatting, as it's buggy upstream.
+        data = result['Stdout'].split()
+        ret = set()
+
+        for ix, line in enumerate(data):
+            # Structure:
+            # $ENTITY
+            # key:
+            # key contents
+            # That's why we need to move one position ahead.
+            if 'key:' in line and entity_filter(data[ix - 1]):
+                ret.add((data[ix - 1], data[ix + 1]))
+        return ret
+
+    def _check_key_rotation(self, entity, unit):
+        def entity_filter(name):
+            return name.startswith(entity)
+
+        old_keys = self._get_all_keys(unit, entity_filter)
+        action_obj = zaza_model.run_action(
+            unit_name=unit,
+            action_name='rotate-key',
+            action_params={'entity': entity}
+        )
+        zaza_utils.assertActionRanOK(action_obj)
+        zaza_model.wait_for_application_states()
+        new_keys = self._get_all_keys(unit, entity_filter)
+        self.assertNotEqual(old_keys, new_keys)
+        diff = new_keys - old_keys
+        self.assertEqual(len(diff), 1)
+        first = next(iter(diff))
+        # Check that the entity matches. The 'entity_filter'
+        # callable will return a true-like value if it
+        # matches the type of entity we're after (i.e: 'mgr')
+        self.assertTrue(entity_filter(first[0]))
+
+    def _get_rgw_client(self, unit):
+        ret = self._get_all_keys(unit, lambda x: x.startswith('client.rgw'))
+        if not ret:
+            return None
+        return next(iter(ret))[0]
+
+    def test_key_rotate(self):
+        """Test that rotating the keys actually changes them."""
+        unit = 'ceph-mon/0'
+        self._check_key_rotation('mgr', unit)
+
+        try:
+            zaza_model.get_application('ceph-radosgw')
+            rgw_client = self._get_rgw_client(unit)
+            if rgw_client:
+                self._check_key_rotation(rgw_client, unit)
+            else:
+                logging.info('ceph-radosgw units present, but no RGW service')
+        except KeyError:
+            pass