Refactor: use mounted from ironic-lib

Change-Id: I0b597ddbc71c133abe6c0acfd8f49e3af4e896bb
This commit is contained in:
Dmitry Tantsur 2021-03-22 14:43:56 +01:00
parent e61336602f
commit d622d38da6
2 changed files with 52 additions and 185 deletions

View File

@ -184,60 +184,37 @@ class GetAgentParamsTestCase(ironic_agent_base.IronicAgentTest):
]) ])
@mock.patch.object(utils, '_find_device_by_labels', autospec=True) @mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(shutil, 'rmtree', autospec=True)
@mock.patch.object(tempfile, 'mkdtemp', autospec=True)
@mock.patch.object(utils, '_read_params_from_file', autospec=True) @mock.patch.object(utils, '_read_params_from_file', autospec=True)
@mock.patch.object(os, 'mkdir', autospec=True) @mock.patch.object(ironic_utils, 'mounted', autospec=True)
@mock.patch.object(utils, 'execute', autospec=True) def test__get_vmedia_params(self, mount_mock, read_params_mock, find_mock):
def test__get_vmedia_params(
self, execute_mock, mkdir_mock, read_params_mock,
mkdtemp_mock, rmtree_mock, find_mock):
mkdtemp_mock.return_value = "/tempdir"
find_mock.return_value = '/dev/fake' find_mock.return_value = '/dev/fake'
mount_mock.return_value.__enter__.return_value = '/tempdir'
null_output = ["", ""]
expected_params = {'a': 'b'} expected_params = {'a': 'b'}
read_params_mock.return_value = expected_params read_params_mock.return_value = expected_params
execute_mock.side_effect = [null_output, null_output]
returned_params = utils._get_vmedia_params() returned_params = utils._get_vmedia_params()
execute_mock.assert_any_call('mount', "/dev/fake", "/tempdir") mount_mock.assert_called_once_with('/dev/fake')
read_params_mock.assert_called_once_with("/tempdir/parameters.txt") read_params_mock.assert_called_once_with("/tempdir/parameters.txt")
execute_mock.assert_any_call('umount', "/tempdir")
self.assertEqual(expected_params, returned_params) self.assertEqual(expected_params, returned_params)
mkdtemp_mock.assert_called_once_with()
rmtree_mock.assert_called_once_with("/tempdir")
@mock.patch.object(utils, '_find_device_by_labels', autospec=True) @mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(shutil, 'rmtree', autospec=True)
@mock.patch.object(tempfile, 'mkdtemp', autospec=True)
@mock.patch.object(utils, '_get_vmedia_device', autospec=True) @mock.patch.object(utils, '_get_vmedia_device', autospec=True)
@mock.patch.object(utils, '_read_params_from_file', autospec=True) @mock.patch.object(utils, '_read_params_from_file', autospec=True)
@mock.patch.object(os, 'mkdir', autospec=True) @mock.patch.object(ironic_utils, 'mounted', autospec=True)
@mock.patch.object(utils, 'execute', autospec=True) def test__get_vmedia_params_by_device(self, mount_mock, read_params_mock,
def test__get_vmedia_params_by_device(self, execute_mock, mkdir_mock, get_device_mock, find_mock):
read_params_mock, get_device_mock,
mkdtemp_mock, rmtree_mock,
find_mock):
mkdtemp_mock.return_value = "/tempdir"
find_mock.return_value = None find_mock.return_value = None
mount_mock.return_value.__enter__.return_value = '/tempdir'
null_output = ["", ""]
expected_params = {'a': 'b'} expected_params = {'a': 'b'}
read_params_mock.return_value = expected_params read_params_mock.return_value = expected_params
execute_mock.side_effect = [null_output, null_output]
get_device_mock.return_value = "sda" get_device_mock.return_value = "sda"
returned_params = utils._get_vmedia_params() returned_params = utils._get_vmedia_params()
execute_mock.assert_any_call('mount', "/dev/sda", mount_mock.assert_called_once_with('/dev/sda')
"/tempdir")
read_params_mock.assert_called_once_with("/tempdir/parameters.txt") read_params_mock.assert_called_once_with("/tempdir/parameters.txt")
execute_mock.assert_any_call('umount', "/tempdir")
self.assertEqual(expected_params, returned_params) self.assertEqual(expected_params, returned_params)
mkdtemp_mock.assert_called_once_with()
rmtree_mock.assert_called_once_with("/tempdir")
@mock.patch.object(utils, '_find_device_by_labels', autospec=True) @mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(utils, '_get_vmedia_device', autospec=True) @mock.patch.object(utils, '_get_vmedia_device', autospec=True)
@ -248,87 +225,6 @@ class GetAgentParamsTestCase(ironic_agent_base.IronicAgentTest):
self.assertRaises(errors.VirtualMediaBootError, self.assertRaises(errors.VirtualMediaBootError,
utils._get_vmedia_params) utils._get_vmedia_params)
@mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(shutil, 'rmtree', autospec=True)
@mock.patch.object(tempfile, 'mkdtemp', autospec=True)
@mock.patch.object(utils, '_read_params_from_file', autospec=True)
@mock.patch.object(os, 'mkdir', autospec=True)
@mock.patch.object(utils, 'execute', autospec=True)
def test__get_vmedia_params_mount_fails(self, execute_mock,
mkdir_mock, read_params_mock,
mkdtemp_mock, rmtree_mock,
find_mock):
find_mock.return_value = '/dev/fake'
mkdtemp_mock.return_value = "/tempdir"
expected_params = {'a': 'b'}
read_params_mock.return_value = expected_params
execute_mock.side_effect = processutils.ProcessExecutionError()
self.assertRaises(errors.VirtualMediaBootError,
utils._get_vmedia_params)
execute_mock.assert_any_call('mount', "/dev/fake", "/tempdir")
mkdtemp_mock.assert_called_once_with()
rmtree_mock.assert_called_once_with("/tempdir")
@mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(shutil, 'rmtree', autospec=True)
@mock.patch.object(tempfile, 'mkdtemp', autospec=True)
@mock.patch.object(utils, '_read_params_from_file', autospec=True)
@mock.patch.object(os, 'mkdir', autospec=True)
@mock.patch.object(utils, 'execute', autospec=True)
def test__get_vmedia_params_umount_fails(self, execute_mock, mkdir_mock,
read_params_mock, mkdtemp_mock,
rmtree_mock, find_mock):
find_mock.return_value = '/dev/fake'
mkdtemp_mock.return_value = "/tempdir"
null_output = ["", ""]
expected_params = {'a': 'b'}
read_params_mock.return_value = expected_params
execute_mock.side_effect = [null_output,
processutils.ProcessExecutionError()]
returned_params = utils._get_vmedia_params()
execute_mock.assert_any_call('mount', "/dev/fake", "/tempdir")
read_params_mock.assert_called_once_with("/tempdir/parameters.txt")
execute_mock.assert_any_call('umount', "/tempdir")
self.assertEqual(expected_params, returned_params)
mkdtemp_mock.assert_called_once_with()
rmtree_mock.assert_called_once_with("/tempdir")
@mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(shutil, 'rmtree', autospec=True)
@mock.patch.object(tempfile, 'mkdtemp', autospec=True)
@mock.patch.object(utils, '_read_params_from_file', autospec=True)
@mock.patch.object(os, 'mkdir', autospec=True)
@mock.patch.object(utils, 'execute', autospec=True)
def test__get_vmedia_params_rmtree_fails(self, execute_mock, mkdir_mock,
read_params_mock, mkdtemp_mock,
rmtree_mock, find_mock):
find_mock.return_value = '/dev/fake'
mkdtemp_mock.return_value = "/tempdir"
rmtree_mock.side_effect = Exception
null_output = ["", ""]
expected_params = {'a': 'b'}
read_params_mock.return_value = expected_params
execute_mock.return_value = null_output
returned_params = utils._get_vmedia_params()
execute_mock.assert_any_call('mount', "/dev/fake", "/tempdir")
read_params_mock.assert_called_once_with("/tempdir/parameters.txt")
execute_mock.assert_any_call('umount', "/tempdir")
self.assertEqual(expected_params, returned_params)
mkdtemp_mock.assert_called_once_with()
rmtree_mock.assert_called_once_with("/tempdir")
class TestFailures(testtools.TestCase): class TestFailures(testtools.TestCase):
def test_get_error(self): def test_get_error(self):
@ -1019,68 +915,68 @@ class TestGetEfiPart(testtools.TestCase):
@mock.patch.object(utils, '_find_device_by_labels', autospec=True) @mock.patch.object(utils, '_find_device_by_labels', autospec=True)
@mock.patch.object(shutil, 'copy', autospec=True) @mock.patch.object(shutil, 'copy', autospec=True)
@mock.patch.object(ironic_utils, 'mounted', autospec=True)
@mock.patch.object(utils, 'execute', autospec=True) @mock.patch.object(utils, 'execute', autospec=True)
class TestCopyConfigFromVmedia(testtools.TestCase): class TestCopyConfigFromVmedia(testtools.TestCase):
def test_no_vmedia(self, mock_execute, mock_copy, mock_find_device): def test_no_vmedia(self, mock_execute, mock_mount, mock_copy,
mock_find_device):
mock_find_device.return_value = None mock_find_device.return_value = None
utils.copy_config_from_vmedia() utils.copy_config_from_vmedia()
mock_mount.assert_not_called()
mock_execute.assert_not_called() mock_execute.assert_not_called()
mock_copy.assert_not_called() mock_copy.assert_not_called()
def test_no_files(self, mock_execute, mock_copy, mock_find_device): def test_no_files(self, mock_execute, mock_mount, mock_copy,
mock_execute.side_effect = [ mock_find_device):
processutils.ProcessExecutionError, temp_path = tempfile.mkdtemp()
('', ''), self.addCleanup(lambda: shutil.rmtree(temp_path))
('', ''),
] mock_execute.side_effect = processutils.ProcessExecutionError
mock_find_device.return_value = '/dev/something' mock_find_device.return_value = '/dev/something'
mock_mount.return_value.__enter__.return_value = temp_path
utils.copy_config_from_vmedia() utils.copy_config_from_vmedia()
mock_execute.assert_has_calls([ mock_mount.assert_called_once_with('/dev/something')
mock.call('findmnt', '-n', '-oTARGET', '/dev/something'), mock_execute.assert_called_once_with('findmnt', '-n', '-oTARGET',
mock.call('mount', '/dev/something', mock.ANY), '/dev/something')
mock.call('umount', mock.ANY),
])
mock_copy.assert_not_called() mock_copy.assert_not_called()
def test_mounted_no_files(self, mock_execute, mock_copy, mock_find_device): def test_mounted_no_files(self, mock_execute, mock_mount, mock_copy,
mock_find_device):
mock_execute.return_value = '/some/path', '' mock_execute.return_value = '/some/path', ''
mock_find_device.return_value = '/dev/something' mock_find_device.return_value = '/dev/something'
utils.copy_config_from_vmedia() utils.copy_config_from_vmedia()
mock_execute.assert_called_once_with( mock_execute.assert_called_once_with(
'findmnt', '-n', '-oTARGET', '/dev/something') 'findmnt', '-n', '-oTARGET', '/dev/something')
mock_copy.assert_not_called() mock_copy.assert_not_called()
mock_mount.assert_not_called()
@mock.patch.object(os, 'makedirs', autospec=True) @mock.patch.object(os, 'makedirs', autospec=True)
def test_copy(self, mock_makedirs, mock_execute, mock_copy, def test_copy(self, mock_makedirs, mock_execute, mock_mount, mock_copy,
mock_find_device): mock_find_device):
mock_find_device.return_value = '/dev/something' mock_find_device.return_value = '/dev/something'
path = None mock_execute.side_effect = processutils.ProcessExecutionError("")
path = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(path))
def _fake_exec(command, arg1, arg2=None, *args): def _fake_mount(dev):
nonlocal path self.assertEqual('/dev/something', dev)
if command == 'mount': # NOTE(dtantsur): makedirs is mocked
path = arg2 os.mkdir(os.path.join(path, 'etc'))
self.assertTrue(os.path.isdir(path)) os.mkdir(os.path.join(path, 'etc', 'ironic-python-agent'))
# NOTE(dtantsur): makedirs is mocked os.mkdir(os.path.join(path, 'etc', 'ironic-python-agent.d'))
os.mkdir(os.path.join(path, 'etc')) with open(os.path.join(path, 'not copied'), 'wt') as fp:
os.mkdir(os.path.join(path, 'etc', 'ironic-python-agent')) fp.write('not copied')
os.mkdir(os.path.join(path, 'etc', 'ironic-python-agent.d')) with open(os.path.join(path, 'etc', 'ironic-python-agent',
with open(os.path.join(path, 'not copied'), 'wt') as fp: 'ironic.crt'), 'wt') as fp:
fp.write('not copied') fp.write('I am a cert')
with open(os.path.join(path, 'etc', 'ironic-python-agent', with open(os.path.join(path, 'etc', 'ironic-python-agent.d',
'ironic.crt'), 'wt') as fp: 'ironic.conf'), 'wt') as fp:
fp.write('I am a cert') fp.write('I am a config')
with open(os.path.join(path, 'etc', 'ironic-python-agent.d', return mock.MagicMock(**{'__enter__.return_value': path})
'ironic.conf'), 'wt') as fp:
fp.write('I am a config')
elif command == 'findmnt':
raise processutils.ProcessExecutionError("")
else:
self.assertEqual('umount', command)
mock_find_device.return_value = '/dev/something' mock_find_device.return_value = '/dev/something'
mock_execute.side_effect = _fake_exec mock_mount.side_effect = _fake_mount
utils.copy_config_from_vmedia() utils.copy_config_from_vmedia()
@ -1088,19 +984,15 @@ class TestCopyConfigFromVmedia(testtools.TestCase):
mock.call('/etc/ironic-python-agent', exist_ok=True), mock.call('/etc/ironic-python-agent', exist_ok=True),
mock.call('/etc/ironic-python-agent.d', exist_ok=True), mock.call('/etc/ironic-python-agent.d', exist_ok=True),
], any_order=True) ], any_order=True)
mock_execute.assert_has_calls([ mock_mount.assert_called_once_with('/dev/something')
mock.call('mount', '/dev/something', mock.ANY),
mock.call('umount', mock.ANY),
])
mock_copy.assert_has_calls([ mock_copy.assert_has_calls([
mock.call(mock.ANY, '/etc/ironic-python-agent/ironic.crt'), mock.call(mock.ANY, '/etc/ironic-python-agent/ironic.crt'),
mock.call(mock.ANY, '/etc/ironic-python-agent.d/ironic.conf'), mock.call(mock.ANY, '/etc/ironic-python-agent.d/ironic.conf'),
], any_order=True) ], any_order=True)
self.assertFalse(os.path.exists(path))
@mock.patch.object(os, 'makedirs', autospec=True) @mock.patch.object(os, 'makedirs', autospec=True)
def test_copy_mounted(self, mock_makedirs, mock_execute, mock_copy, def test_copy_mounted(self, mock_makedirs, mock_execute, mock_mount,
mock_find_device): mock_copy, mock_find_device):
mock_find_device.return_value = '/dev/something' mock_find_device.return_value = '/dev/something'
path = tempfile.mkdtemp() path = tempfile.mkdtemp()
self.addCleanup(lambda: shutil.rmtree(path)) self.addCleanup(lambda: shutil.rmtree(path))
@ -1133,6 +1025,7 @@ class TestCopyConfigFromVmedia(testtools.TestCase):
mock.call(mock.ANY, '/etc/ironic-python-agent/ironic.crt'), mock.call(mock.ANY, '/etc/ironic-python-agent/ironic.crt'),
mock.call(mock.ANY, '/etc/ironic-python-agent.d/ironic.conf'), mock.call(mock.ANY, '/etc/ironic-python-agent.d/ironic.conf'),
], any_order=True) ], any_order=True)
mock_mount.assert_not_called()
@mock.patch.object(requests, 'get', autospec=True) @mock.patch.object(requests, 'get', autospec=True)

View File

@ -24,7 +24,6 @@ import shutil
import subprocess import subprocess
import sys import sys
import tarfile import tarfile
import tempfile
import time import time
from ironic_lib import disk_utils from ironic_lib import disk_utils
@ -126,31 +125,6 @@ def _get_vmedia_device():
pass pass
@contextlib.contextmanager
def _mounted(source):
"""A context manager for a temporary mount."""
dest = tempfile.mkdtemp()
try:
try:
execute("mount", source, dest)
except processutils.ProcessExecutionError as e:
msg = ("Unable to mount virtual media device %(device)s: "
"%(error)s" % {'device': source, 'error': e})
raise errors.VirtualMediaBootError(msg)
yield dest
finally:
try:
execute("umount", dest)
except processutils.ProcessExecutionError:
pass
try:
shutil.rmtree(dest)
except Exception:
pass
def _find_device_by_labels(labels): def _find_device_by_labels(labels):
"""Find device matching any of the provided labels.""" """Find device matching any of the provided labels."""
for label in labels + [lbl.upper() for lbl in labels]: for label in labels + [lbl.upper() for lbl in labels]:
@ -181,7 +155,7 @@ def _get_vmedia_params():
vmedia_device_file = os.path.join("/dev", vmedia_device) vmedia_device_file = os.path.join("/dev", vmedia_device)
with _mounted(vmedia_device_file) as vmedia_mount_point: with ironic_utils.mounted(vmedia_device_file) as vmedia_mount_point:
parameters_file_path = os.path.join(vmedia_mount_point, parameters_file_path = os.path.join(vmedia_mount_point,
parameters_file) parameters_file)
params = _read_params_from_file(parameters_file_path) params = _read_params_from_file(parameters_file_path)
@ -242,7 +216,7 @@ def copy_config_from_vmedia():
if mounted: if mounted:
_copy_config_from(mounted) _copy_config_from(mounted)
else: else:
with _mounted(vmedia_device_file) as vmedia_mount_point: with ironic_utils.mounted(vmedia_device_file) as vmedia_mount_point:
_copy_config_from(vmedia_mount_point) _copy_config_from(vmedia_mount_point)