From 9881001be32c4d8a8496302bd7e7a854f45413e5 Mon Sep 17 00:00:00 2001 From: Chris MacNaughton Date: Tue, 14 Feb 2017 09:13:55 -0600 Subject: [PATCH] Fix get_named_key this change allows the get_named_key to check if a key with the given name already exists (with any permissions) and returns the named key if so. Otherwise, it returns a new key created with the requested name and permissions Closes-Bug: 1664546 Change-Id: Ifd1ff35e5e55e2f4d480609643174d3bf4afd0c0 --- ceph/__init__.py | 20 ++++++++++++ unit_tests/test_ceph.py | 68 +++++++++++++++++++++++++---------------- 2 files changed, 62 insertions(+), 26 deletions(-) diff --git a/ceph/__init__.py b/ceph/__init__.py index e87aef9..26f1b91 100644 --- a/ceph/__init__.py +++ b/ceph/__init__.py @@ -1092,6 +1092,26 @@ def get_named_key(name, caps=None, pool_list=None): :param caps: dict of cephx capabilities :return: Returns a cephx key """ + try: + # Does the key already exist? + output = subprocess.check_output( + [ + 'sudo', + '-u', ceph_user(), + 'ceph', + '--name', 'mon.', + '--keyring', + '/var/lib/ceph/mon/ceph-{}/keyring'.format( + socket.gethostname() + ), + 'auth', + 'get', + 'client.{}'.format(name), + ]).strip() + return parse_key(output) + except subprocess.CalledProcessError: + # Couldn't get the key, time to create it! + log("Creating new key for {}".format(name), level=DEBUG) caps = caps or _default_caps cmd = [ "sudo", diff --git a/unit_tests/test_ceph.py b/unit_tests/test_ceph.py index 9d67a2d..dde1147 100644 --- a/unit_tests/test_ceph.py +++ b/unit_tests/test_ceph.py @@ -15,9 +15,11 @@ import mock import unittest import ceph +from subprocess import CalledProcessError class TestDevice(): + """Test class to mock out pyudev Device""" def __getitem__(**kwargs): @@ -34,6 +36,7 @@ class TestDevice(): class CephTestCase(unittest.TestCase): + def setUp(self): super(CephTestCase, self).setUp() @@ -110,32 +113,45 @@ class CephTestCase(unittest.TestCase): as subprocess: with mock.patch.object(ceph.socket, "gethostname", return_value="osd001"): + subprocess.side_effect = [ + CalledProcessError(0, 0, 0), ""] ceph.get_named_key(name="rgw001", pool_list=["rbd", "block"]) - subprocess.assert_called_with( - ['sudo', '-u', 'ceph', 'ceph', '--name', 'mon.', - '--keyring', - '/var/lib/ceph/mon/ceph-osd001/keyring', - 'auth', - 'get-or-create', 'client.rgw001', 'mon', 'allow r', - 'osd', - 'allow rwx pool=rbd pool=block']) + subprocess.assert_has_calls([ + mock.call(['sudo', '-u', 'ceph', 'ceph', '--name', + 'mon.', '--keyring', + '/var/lib/ceph/mon/ceph-osd001/keyring', + 'auth', 'get', 'client.rgw001']), + mock.call(['sudo', '-u', 'ceph', 'ceph', '--name', + 'mon.', '--keyring', + '/var/lib/ceph/mon/ceph-osd001/keyring', + 'auth', 'get-or-create', 'client.rgw001', + 'mon', 'allow r', 'osd', + 'allow rwx pool=rbd pool=block'])]) def test_get_named_key(self): with mock.patch.object(ceph, "ceph_user", return_value="ceph"): with mock.patch.object(ceph.subprocess, "check_output") \ as subprocess: + subprocess.side_effect = [ + CalledProcessError(0, 0, 0), + ""] with mock.patch.object(ceph.socket, "gethostname", return_value="osd001"): ceph.get_named_key(name="rgw001") - subprocess.assert_called_with( - ['sudo', '-u', 'ceph', 'ceph', '--name', 'mon.', - '--keyring', - '/var/lib/ceph/mon/ceph-osd001/keyring', - 'auth', - 'get-or-create', 'client.rgw001', 'mon', 'allow r', - 'osd', - 'allow rwx']) + for call in subprocess.mock_calls: + print("Subprocess: {}".format(call)) + subprocess.assert_has_calls([ + mock.call(['sudo', '-u', 'ceph', 'ceph', '--name', + 'mon.', '--keyring', + '/var/lib/ceph/mon/ceph-osd001/keyring', + 'auth', 'get', 'client.rgw001']), + mock.call(['sudo', '-u', 'ceph', 'ceph', '--name', + 'mon.', '--keyring', + '/var/lib/ceph/mon/ceph-osd001/keyring', + 'auth', 'get-or-create', 'client.rgw001', + 'mon', 'allow r', 'osd', + 'allow rwx'])]) def test_list_unmounted_devices(self): dev1 = mock.MagicMock(spec=TestDevice) @@ -151,16 +167,16 @@ class CephTestCase(unittest.TestCase): with mock.patch( 'pyudev.Context.list_devices', return_value=devices): - with mock.patch.object(ceph, - 'is_device_mounted', - return_value=False): - devices = ceph.unmounted_disks() - self.assertEqual(devices, ['/dev/sda', '/dev/sdb']) - with mock.patch.object(ceph, - 'is_device_mounted', - return_value=True): - devices = ceph.unmounted_disks() - self.assertEqual(devices, []) + with mock.patch.object(ceph, + 'is_device_mounted', + return_value=False): + devices = ceph.unmounted_disks() + self.assertEqual(devices, ['/dev/sda', '/dev/sdb']) + with mock.patch.object(ceph, + 'is_device_mounted', + return_value=True): + devices = ceph.unmounted_disks() + self.assertEqual(devices, []) class CephVersionTestCase(unittest.TestCase):