diff --git a/zaza/openstack/charm_tests/ceph/tests.py b/zaza/openstack/charm_tests/ceph/tests.py index 0beac49..85de53d 100644 --- a/zaza/openstack/charm_tests/ceph/tests.py +++ b/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