openstack/ironic-python-agent
Code Issues Proposed changes

Merge "Refactor list_all_block_devices & add block_type param"

Browse Source
This commit is contained in:
Jenkins 2015-10-07 02:42:06 +00:00 committed by Gerrit Code Review
commit aa908205c6
2 changed files with 59 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
ironic_python_agent/hardware.py
View File

@ -49,19 +49,22 @@ def _get_device_vendor(dev):
LOG.warning("Can't find the device vendor for device %s", dev)
def list_all_block_devices():
def list_all_block_devices(block_type='disk'):
"""List all physical block devices
The switches we use for lsblk: P for KEY="value" output, b for size output
in bytes, d to exclude dependent devices (like md or dm devices), i to
ensure ascii characters only, and o to specify the fields we need.
ensure ascii characters only, and o to specify the fields/columns we need.
Broken out as its own function to facilitate custom hardware managers that
don't need to subclass GenericHardwareManager.
:param block_type: Type of block device to find
:return: A list of BlockDevices
"""
report = utils.execute('lsblk', '-PbdioKNAME,MODEL,SIZE,ROTA,TYPE',
columns = ['KNAME', 'MODEL', 'SIZE', 'ROTA', 'TYPE']
report = utils.execute('lsblk', '-Pbdi', '-o{}'.format(','.join(columns)),
check_exit_code=[0])[0]
lines = report.split('\n')
context = pyudev.Context()
@ -73,15 +76,15 @@ def list_all_block_devices():
vals = shlex.split(line)
for key, val in (v.split('=', 1) for v in vals):
device[key] = val.strip()
# Ignore non disk
if device.get('TYPE') != 'disk':
# Ignore block types not specified
if device.get('TYPE') != block_type:
continue
# Ensure all required keys are at least present, even if blank
diff = set(['KNAME', 'MODEL', 'SIZE', 'ROTA']) - set(device.keys())
if diff:
# Ensure all required columns are at least present, even if blank
missing = set(columns) - set(device)
if missing:
raise errors.BlockDeviceError(
'%s must be returned by lsblk.' % diff)
'%s must be returned by lsblk.' % ', '.join(sorted(missing)))
name = '/dev/' + device['KNAME']
try:

49
ironic_python_agent/tests/unit/test_hardware.py
View File

@ -140,6 +140,14 @@ BLK_DEVICE_TEMPLATE_SMALL = (
'KNAME="sdb" MODEL="AlmostBigEnough Drive" SIZE="4294967295" '
'ROTA="0" TYPE="disk"'
)
BLK_DEVICE_TEMPLATE_SMALL_DEVICES = [
hardware.BlockDevice(name='/dev/sda', model='TinyUSB Drive',
size=3116853504, rotational=False,
vendor="FooTastic"),
hardware.BlockDevice(name='/dev/sdb', model='AlmostBigEnough Drive',
size=4294967295, rotational=False,
vendor="FooTastic"),
]
SHRED_OUTPUT = (
'shred: /dev/sda: pass 1/2 (random)...\n'
@ -280,7 +288,7 @@ class TestGenericHardwareManager(test_base.BaseTestCase):
mocked_execute.return_value = (BLK_DEVICE_TEMPLATE, '')
self.assertEqual(self.hardware.get_os_install_device(), '/dev/sdb')
mocked_execute.assert_called_once_with(
'lsblk', '-PbdioKNAME,MODEL,SIZE,ROTA,TYPE',
'lsblk', '-Pbdi', '-oKNAME,MODEL,SIZE,ROTA,TYPE',
check_exit_code=[0])
@mock.patch.object(utils, 'execute')
@ -290,7 +298,8 @@ class TestGenericHardwareManager(test_base.BaseTestCase):
ex = self.assertRaises(errors.DeviceNotFound,
self.hardware.get_os_install_device)
mocked_execute.assert_called_once_with(
'lsblk', '-PbdioKNAME,MODEL,SIZE,ROTA,TYPE', check_exit_code=[0])
'lsblk', '-Pbdi', '-oKNAME,MODEL,SIZE,ROTA,TYPE',
check_exit_code=[0])
self.assertIn(str(4 * units.Gi), ex.details)
@mock.patch.object(hardware, 'list_all_block_devices')
@ -797,3 +806,39 @@ class TestGenericHardwareManager(test_base.BaseTestCase):
def test_get_bmc_address_virt(self, mocked_execute):
mocked_execute.side_effect = processutils.ProcessExecutionError()
self.assertIsNone(self.hardware.get_bmc_address())
class TestModuleFunctions(test_base.BaseTestCase):
@mock.patch.object(hardware, '_get_device_vendor', lambda x: "FooTastic")
@mock.patch.object(hardware.pyudev.Device, "from_device_file")
@mock.patch.object(utils, 'execute', autospec=True)
def test_list_all_block_devices_success(self, mocked_execute,
mocked_fromdevfile):
mocked_fromdevfile.return_value = {}
mocked_execute.return_value = (BLK_DEVICE_TEMPLATE_SMALL, '')
result = hardware.list_all_block_devices()
mocked_execute.assert_called_once_with(
'lsblk', '-Pbdi', '-oKNAME,MODEL,SIZE,ROTA,TYPE',
check_exit_code=[0])
self.assertEqual(BLK_DEVICE_TEMPLATE_SMALL_DEVICES, result)
@mock.patch.object(utils, 'execute', autospec=True)
@mock.patch.object(hardware, '_get_device_vendor', lambda x: "FooTastic")
def test_list_all_block_devices_wrong_block_type(self, mocked_execute):
mocked_execute.return_value = ('TYPE="foo" MODEL="model"', '')
result = hardware.list_all_block_devices()
mocked_execute.assert_called_once_with(
'lsblk', '-Pbdi', '-oKNAME,MODEL,SIZE,ROTA,TYPE',
check_exit_code=[0])
self.assertEqual([], result)
@mock.patch.object(utils, 'execute', autospec=True)
def test_list_all_block_devices_missing(self, mocked_execute):
"""Test for missing values returned from lsblk"""
mocked_execute.return_value = ('TYPE="disk" MODEL="model"', '')
self.assertRaisesRegexp(
errors.BlockDeviceError,
r'^Block device caused unknown error: KNAME, ROTA, SIZE must be '
r'returned by lsblk.$',
hardware.list_all_block_devices)