openstack/os-brick
Code Issues Proposed changes

Merge "RBD: ensure temporary config gets deleted"

Browse Source
This commit is contained in:
Jenkins 2016-12-07 19:13:41 +00:00 committed by Gerrit Code Review
commit 902569547f
2 changed files with 11 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
os_brick/initiator/connectors/rbd.py
View File

@ -18,6 +18,7 @@ import tempfile
from oslo_concurrency import processutils as putils
from oslo_log import log as logging
from oslo_utils import fileutils
from oslo_utils import netutils
from os_brick.i18n import _, _LE
@ -78,7 +79,7 @@ class RBDConnector(base.BaseLinuxConnector):
keyring = ("keyring = /etc/ceph/%s.client.%s.keyring" %
(cluster_name, user))
try:
fd, ceph_conf_path = tempfile.mkstemp()
fd, ceph_conf_path = tempfile.mkstemp(prefix="brickrbd_")
with os.fdopen(fd, 'w') as conf_file:
conf_file.writelines([mon_hosts, "\n",
client_section, "\n", keyring])
@ -100,14 +101,14 @@ class RBDConnector(base.BaseLinuxConnector):
conf = self._create_ceph_conf(monitor_ips, monitor_ports,
str(cluster_name), user)
rbd_client = linuxrbd.RBDClient(user, pool, conffile=conf,
rbd_cluster_name=str(cluster_name))
rbd_volume = linuxrbd.RBDVolume(rbd_client, volume)
rbd_handle = linuxrbd.RBDVolumeIOWrapper(
linuxrbd.RBDImageMetadata(rbd_volume, pool, user, conf))
if os.path.exists(conf):
os.remove(conf)
try:
rbd_client = linuxrbd.RBDClient(user, pool, conffile=conf,
rbd_cluster_name=str(cluster_name))
rbd_volume = linuxrbd.RBDVolume(rbd_client, volume)
rbd_handle = linuxrbd.RBDVolumeIOWrapper(
linuxrbd.RBDImageMetadata(rbd_volume, pool, user, conf))
finally:
fileutils.delete_if_exists(conf)
return rbd_handle

2
os_brick/tests/initiator/connectors/test_rbd.py
View File

@ -124,7 +124,7 @@ class RBDConnectorTestCase(test_connector.ConnectorTestCase):
conf_path = rbd_connector._create_ceph_conf(
self.hosts, self.ports, self.clustername, self.user)
self.assertEqual(conf_path, tmpfile)
mock_mkstemp.assert_called_once_with()
mock_mkstemp.assert_called_once_with(prefix='brickrbd_')
@mock.patch.object(priv_rootwrap, 'execute', return_value=None)
def test_connect_local_volume(self, mock_execute):