Merge "Extend parse_root_device_hints to support operators"

This commit is contained in:
Jenkins 2016-08-30 18:06:59 +00:00 committed by Gerrit Code Review
commit d4ed5a9d96
2 changed files with 287 additions and 37 deletions

View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import errno
import os
import os.path
@ -283,31 +284,68 @@ class IsHttpUrlTestCase(test_base.BaseTestCase):
class ParseRootDeviceTestCase(test_base.BaseTestCase):
def setUp(self):
super(ParseRootDeviceTestCase, self).setUp()
self.root_device = {
'wwn': '123456', 'model': 'foo-model', 'size': 12345,
'serial': 'foo-serial', 'vendor': 'foo-vendor', 'name': '/dev/sda',
'wwn_with_extension': '123456111', 'wwn_vendor_extension': '111',
'rotational': True}
def test_parse_root_device_hints_without_operators(self):
root_device = {
'wwn': '123456', 'model': 'FOO model', 'size': 12345,
'serial': 'foo-serial', 'vendor': 'foo VENDOR with space',
'name': '/dev/sda', 'wwn_with_extension': '123456111',
'wwn_vendor_extension': '111', 'rotational': True}
result = utils.parse_root_device_hints(root_device)
expected = {
'wwn': 's== 123456', 'model': 's== foo%20model',
'size': '== 12345', 'serial': 's== foo-serial',
'vendor': 's== foo%20vendor%20with%20space',
'name': 's== /dev/sda', 'wwn_with_extension': 's== 123456111',
'wwn_vendor_extension': 's== 111', 'rotational': True}
self.assertEqual(expected, result)
def test_parse_root_device_hints(self):
result = utils.parse_root_device_hints(self.root_device)
self.assertEqual(self.root_device, result)
def test_parse_root_device_hints_with_operators(self):
root_device = {
'wwn': 's== 123456', 'model': 's== foo MODEL', 'size': '>= 12345',
'serial': 's!= foo-serial', 'vendor': 's== foo VENDOR with space',
'name': '<or> /dev/sda <or> /dev/sdb',
'wwn_with_extension': 's!= 123456111',
'wwn_vendor_extension': 's== 111', 'rotational': True}
# Validate strings being normalized
expected = copy.deepcopy(root_device)
expected['model'] = 's== foo%20model'
expected['vendor'] = 's== foo%20vendor%20with%20space'
result = utils.parse_root_device_hints(root_device)
# The hints already contain the operators, make sure we keep it
self.assertEqual(expected, result)
def test_parse_root_device_hints_no_hints(self):
result = utils.parse_root_device_hints({})
self.assertIsNone(result)
def test_parse_root_device_hints_convert_size(self):
result = utils.parse_root_device_hints({'size': '12345'})
self.assertEqual({'size': 12345}, result)
for size in (12345, '12345'):
result = utils.parse_root_device_hints({'size': size})
self.assertEqual({'size': '== 12345'}, result)
def test_parse_root_device_hints_invalid_size(self):
for value in ('not-int', -123, 0):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'size': value})
def test_parse_root_device_hints_int_or(self):
expr = '<or> 123 <or> 456 <or> 789'
result = utils.parse_root_device_hints({'size': expr})
self.assertEqual({'size': expr}, result)
def test_parse_root_device_hints_int_or_invalid(self):
expr = '<or> 123 <or> non-int <or> 789'
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'size': expr})
def test_parse_root_device_hints_string_or_space(self):
expr = '<or> foo <or> foo bar <or> bar'
expected = '<or> foo <or> foo%20bar <or> bar'
result = utils.parse_root_device_hints({'model': expr})
self.assertEqual({'model': expected}, result)
def _parse_root_device_hints_convert_rotational(self, values,
expected_value):
for value in values:
@ -325,6 +363,115 @@ class ParseRootDeviceTestCase(test_base.BaseTestCase):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'rotational': 'not-bool'})
def test_parse_root_device_hints_invalid_wwn(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'wwn': 123})
def test_parse_root_device_hints_invalid_wwn_with_extension(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'wwn_with_extension': 123})
def test_parse_root_device_hints_invalid_wwn_vendor_extension(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'wwn_vendor_extension': 123})
def test_parse_root_device_hints_invalid_model(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'model': 123})
def test_parse_root_device_hints_invalid_serial(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'serial': 123})
def test_parse_root_device_hints_invalid_vendor(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'vendor': 123})
def test_parse_root_device_hints_invalid_name(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'name': 123})
def test_parse_root_device_hints_non_existent_hint(self):
self.assertRaises(ValueError, utils.parse_root_device_hints,
{'non-existent': 'foo'})
def test_extract_hint_operator_and_values_single_value(self):
expected = {'op': '>=', 'values': ['123']}
self.assertEqual(
expected, utils._extract_hint_operator_and_values(
'>= 123', 'size'))
def test_extract_hint_operator_and_values_multiple_values(self):
expected = {'op': '<or>', 'values': ['123', '456', '789']}
expr = '<or> 123 <or> 456 <or> 789'
self.assertEqual(
expected, utils._extract_hint_operator_and_values(expr, 'size'))
def test_extract_hint_operator_and_values_multiple_values_space(self):
expected = {'op': '<or>', 'values': ['foo', 'foo bar', 'bar']}
expr = '<or> foo <or> foo bar <or> bar'
self.assertEqual(
expected, utils._extract_hint_operator_and_values(expr, 'model'))
def test_extract_hint_operator_and_values_no_operator(self):
expected = {'op': '', 'values': ['123']}
self.assertEqual(
expected, utils._extract_hint_operator_and_values('123', 'size'))
def test_extract_hint_operator_and_values_empty_value(self):
self.assertRaises(
ValueError, utils._extract_hint_operator_and_values, '', 'size')
def test_extract_hint_operator_and_values_integer(self):
expected = {'op': '', 'values': ['123']}
self.assertEqual(
expected, utils._extract_hint_operator_and_values(123, 'size'))
def test__append_operator_to_hints(self):
root_device = {'serial': 'foo', 'size': 12345,
'model': 'foo model', 'rotational': True}
expected = {'serial': 's== foo', 'size': '== 12345',
'model': 's== foo model', 'rotational': True}
result = utils._append_operator_to_hints(root_device)
self.assertEqual(expected, result)
def test_normalize_hint_expression_or(self):
expr = '<or> foo <or> foo bar <or> bar'
expected = '<or> foo <or> foo%20bar <or> bar'
result = utils._normalize_hint_expression(expr, 'model')
self.assertEqual(expected, result)
def test_normalize_hint_expression_in(self):
expr = '<in> foo <in> foo bar <in> bar'
expected = '<in> foo <in> foo%20bar <in> bar'
result = utils._normalize_hint_expression(expr, 'model')
self.assertEqual(expected, result)
def test_normalize_hint_expression_op_space(self):
expr = 's== test string with space'
expected = 's== test%20string%20with%20space'
result = utils._normalize_hint_expression(expr, 'model')
self.assertEqual(expected, result)
def test_normalize_hint_expression_op_no_space(self):
expr = 's!= SpongeBob'
expected = 's!= spongebob'
result = utils._normalize_hint_expression(expr, 'model')
self.assertEqual(expected, result)
def test_normalize_hint_expression_no_op_space(self):
expr = 'no operators'
expected = 'no%20operators'
result = utils._normalize_hint_expression(expr, 'model')
self.assertEqual(expected, result)
def test_normalize_hint_expression_no_op_no_space(self):
expr = 'NoSpace'
expected = 'nospace'
result = utils._normalize_hint_expression(expr, 'model')
self.assertEqual(expected, result)
def test_normalize_hint_expression_empty_value(self):
self.assertRaises(
ValueError, utils._normalize_hint_expression, '', 'size')

View File

@ -22,11 +22,15 @@ import copy
import errno
import logging
import os
import re
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import excutils
from oslo_utils import specs_matcher
from oslo_utils import strutils
import six
from six.moves.urllib import parse
from ironic_lib.common.i18n import _
from ironic_lib.common.i18n import _LE
@ -45,10 +49,15 @@ CONF.register_opts(utils_opts, group='ironic_lib')
LOG = logging.getLogger(__name__)
# A dictionary in the form {hint name: hint type}
VALID_ROOT_DEVICE_HINTS = {
'size': int, 'model': str, 'wwn': str, 'serial': str, 'vendor': str,
'wwn_with_extension': str, 'wwn_vendor_extension': str, 'name': str,
'rotational': bool,
}
VALID_ROOT_DEVICE_HINTS = set(('size', 'model', 'wwn', 'serial', 'vendor',
'wwn_with_extension', 'wwn_vendor_extension',
'name', 'rotational'))
ROOT_DEVICE_HINTS_GRAMMAR = specs_matcher.make_grammar()
def execute(*cmd, **kwargs):
@ -169,6 +178,87 @@ def list_opts():
return [('ironic_lib', utils_opts)]
def _extract_hint_operator_and_values(hint_expression, hint_name):
"""Extract the operator and value(s) of a root device hint expression.
A root device hint expression could contain one or more values
depending on the operator. This method extracts the operator and
value(s) and returns a dictionary containing both.
:param hint_expression: The hint expression string containing value(s)
and operator (optionally).
:param hint_name: The name of the hint. Used for logging.
:raises: ValueError if the hint_expression is empty.
:returns: A dictionary containing:
:op: The operator. An empty string in case of None.
:values: A list of values stripped and converted to lowercase.
"""
expression = six.text_type(hint_expression).strip().lower()
if not expression:
raise ValueError(
_('Root device hint "%s" expression is empty') % hint_name)
# parseString() returns a list of tokens which the operator (if
# present) is always the first element.
ast = ROOT_DEVICE_HINTS_GRAMMAR.parseString(expression)
if len(ast) <= 1:
# hint_expression had no operator
return {'op': '', 'values': [expression]}
op = ast[0]
return {'values': [v.strip() for v in re.split(op, expression) if v],
'op': op}
def _normalize_hint_expression(hint_expression, hint_name):
"""Normalize a string type hint expression.
A string-type hint expression contains one or more operators and
one or more values: [<op>] <value> [<op> <value>]*. This normalizes
the values by url-encoding white spaces and special characters. The
operators are not normalized. For example: the hint value of "<or>
foo bar <or> bar" will become "<or> foo%20bar <or> bar".
:param hint_expression: The hint expression string containing value(s)
and operator (optionally).
:param hint_name: The name of the hint. Used for logging.
:raises: ValueError if the hint_expression is empty.
:returns: A normalized string.
"""
hdict = _extract_hint_operator_and_values(hint_expression, hint_name)
result = hdict['op'].join([' %s ' % parse.quote(t)
for t in hdict['values']])
return (hdict['op'] + result).strip()
def _append_operator_to_hints(root_device):
"""Add an equal (s== or ==) operator to the hints.
For backwards compatibility, for root device hints where no operator
means equal, this method adds the equal operator to the hint. This is
needed when using oslo.utils.specs_matcher methods.
:param root_device: The root device hints dictionary.
"""
for name, expression in root_device.items():
# NOTE(lucasagomes): The specs_matcher from oslo.utils does not
# support boolean, so we don't need to append any operator
# for it.
if VALID_ROOT_DEVICE_HINTS[name] is bool:
continue
expression = six.text_type(expression)
ast = ROOT_DEVICE_HINTS_GRAMMAR.parseString(expression)
if len(ast) > 1:
continue
op = 's== %s' if VALID_ROOT_DEVICE_HINTS[name] is str else '== %s'
root_device[name] = op % expression
return root_device
def parse_root_device_hints(root_device):
"""Parse the root_device property of a node.
@ -188,7 +278,7 @@ def parse_root_device_hints(root_device):
root_device = copy.deepcopy(root_device)
invalid_hints = set(root_device) - VALID_ROOT_DEVICE_HINTS
invalid_hints = set(root_device) - set(VALID_ROOT_DEVICE_HINTS)
if invalid_hints:
raise ValueError(
_('The hints "%(invalid_hints)s" are invalid. '
@ -196,28 +286,41 @@ def parse_root_device_hints(root_device):
{'invalid_hints': ', '.join(invalid_hints),
'valid_hints': ', '.join(VALID_ROOT_DEVICE_HINTS)})
if 'size' in root_device:
try:
size = int(root_device['size'])
except ValueError:
raise ValueError(
_('Root device hint "size" is not an integer value. '
'Current value: %s') % root_device['size'])
for name, expression in root_device.items():
hint_type = VALID_ROOT_DEVICE_HINTS[name]
if hint_type is str:
if not isinstance(expression, six.string_types):
raise ValueError(
_('Root device hint "%(name)s" is not a string value. '
'Hint expression: %(expression)s') %
{'name': name, 'expression': expression})
root_device[name] = _normalize_hint_expression(expression, name)
if size <= 0:
raise ValueError(
_('Root device hint "size" should be a positive integer. '
'Current value: %d') % size)
elif hint_type is int:
for v in _extract_hint_operator_and_values(expression,
name)['values']:
try:
integer = int(v)
except ValueError:
raise ValueError(
_('Root device hint "%(name)s" is not an integer '
'value. Current value: %(expression)s') %
{'name': name, 'expression': expression})
root_device['size'] = size
if integer <= 0:
raise ValueError(
_('Root device hint "%(name)s" should be a positive '
'integer. Current value: %(expression)s') %
{'name': name, 'expression': expression})
if 'rotational' in root_device:
try:
root_device['rotational'] = strutils.bool_from_string(
root_device['rotational'], strict=True)
except ValueError:
raise ValueError(
_('Root device hint "rotational" is not a Boolean value. '
'Current value: %s') % root_device['rotational'])
elif hint_type is bool:
try:
root_device[name] = strutils.bool_from_string(
expression, strict=True)
except ValueError:
raise ValueError(
_('Root device hint "%(name)s" is not a Boolean value. '
'Current value: %(expression)s') %
{'name': name, 'expression': expression})
return root_device
return _append_operator_to_hints(root_device)