Merge "Validate arguments to clean_step() decorator"

This commit is contained in:
Jenkins 2015-12-14 23:39:17 +00:00 committed by Gerrit Code Review
commit 3732114d5c
2 changed files with 191 additions and 6 deletions

View File

@ -980,6 +980,60 @@ class RAIDInterface(BaseInterface):
return raid.get_logical_disk_properties(self.raid_schema)
def _validate_argsinfo(argsinfo):
"""Validate args info.
This method validates args info, so that the values are the expected
data types and required values are specified.
:param argsinfo: a dictionary of keyword arguments where key is the name of
the argument and value is a dictionary as follows::
description: <description>. Required. This should include
possible values.
required: Boolean. Optional; default is False. True if this
argument is required. If so, it must be specified in
the clean request; false if it is optional.
:raises InvalidParameterValue if any of the arguments are invalid
"""
if not argsinfo:
return
if not isinstance(argsinfo, dict):
raise exception.InvalidParameterValue(
_('"argsinfo" must be a dictionary instead of "%s"') %
argsinfo)
for (arg, info) in argsinfo.items():
if not isinstance(info, dict):
raise exception.InvalidParameterValue(
_('Argument "%(arg)s" must be a dictionary instead of '
'"%(val)s".') % {'arg': arg, 'val': info})
has_description = False
for (key, value) in info.items():
if key == 'description':
if not isinstance(value, six.string_types):
raise exception.InvalidParameterValue(
_('For argument "%(arg)s", "description" must be a '
'string value instead of "%(value)s".') %
{'arg': arg, 'value': value})
has_description = True
elif key == 'required':
if not isinstance(value, bool):
raise exception.InvalidParameterValue(
_('For argument "%(arg)s", "required" must be a '
'Boolean value instead of "%(value)s".') %
{'arg': arg, 'value': value})
else:
raise exception.InvalidParameterValue(
_('Argument "%(arg)s" has an invalid key named "%(key)s". '
'It must be "description" or "required".')
% {'key': key, 'arg': arg})
if not has_description:
raise exception.InvalidParameterValue(
_('Argument "%(arg)s" is missing a "description".') %
{'arg': arg})
def clean_step(priority, abortable=False, argsinfo=None):
"""Decorator for cleaning steps.
@ -1029,15 +1083,30 @@ def clean_step(priority, abortable=False, argsinfo=None):
:param argsinfo: a dictionary of keyword arguments where key is the name of
the argument and value is a dictionary as follows::
description: <description>. This should include possible values.
required: Boolean. True if this argument is required. If so, it
must be specified in the clean request; false if it is
optional.
description: <description>. Required. This should include
possible values.
required: Boolean. Optional; default is False. True if this
argument is required. If so, it must be specified in
the clean request; false if it is optional.
:raises InvalidParameterValue if any of the arguments are invalid
"""
def decorator(func):
func._is_clean_step = True
func._clean_step_priority = priority
func._clean_step_abortable = abortable
if isinstance(priority, int):
func._clean_step_priority = priority
else:
raise exception.InvalidParameterValue(
_('"priority" must be an integer value instead of "%s"')
% priority)
if isinstance(abortable, bool):
func._clean_step_abortable = abortable
else:
raise exception.InvalidParameterValue(
_('"abortable" must be a Boolean value instead of "%s"')
% abortable)
_validate_argsinfo(argsinfo)
func._clean_step_argsinfo = argsinfo
return func
return decorator

View File

@ -117,6 +117,122 @@ class DriverPeriodicTaskTestCase(base.TestCase):
self.assertEqual(1, spawn_mock.call_count)
class CleanStepDecoratorTestCase(base.TestCase):
def setUp(self):
super(CleanStepDecoratorTestCase, self).setUp()
method_mock = mock.MagicMock()
del method_mock._is_clean_step
del method_mock._clean_step_priority
del method_mock._clean_step_abortable
del method_mock._clean_step_argsinfo
self.method = method_mock
def test__validate_argsinfo(self):
# None, empty dict
driver_base._validate_argsinfo(None)
driver_base._validate_argsinfo({})
# Only description specified
driver_base._validate_argsinfo({'arg1': {'description': 'desc1'}})
# Multiple args
driver_base._validate_argsinfo({'arg1': {'description': 'desc1',
'required': True},
'arg2': {'description': 'desc2'}})
def test__validate_argsinfo_not_dict(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'argsinfo.+dictionary',
driver_base._validate_argsinfo, 'not-a-dict')
def test__validate_argsinfo_arg_not_dict(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'Argument.+dictionary',
driver_base._validate_argsinfo,
{'arg1': 'not-a-dict'})
def test__validate_argsinfo_arg_empty_dict(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'description',
driver_base._validate_argsinfo,
{'arg1': {}})
def test__validate_argsinfo_arg_missing_description(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'description',
driver_base._validate_argsinfo,
{'arg1': {'required': True}})
def test__validate_argsinfo_arg_description_invalid(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'string',
driver_base._validate_argsinfo,
{'arg1': {'description': True}})
def test__validate_argsinfo_arg_required_invalid(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'Boolean',
driver_base._validate_argsinfo,
{'arg1': {'description': 'desc1',
'required': 'maybe'}})
def test__validate_argsinfo_arg_unknown_key(self):
self.assertRaisesRegexp(exception.InvalidParameterValue,
'invalid',
driver_base._validate_argsinfo,
{'arg1': {'description': 'desc1',
'unknown': 'bad'}})
def test_clean_step_priority_only(self):
d = driver_base.clean_step(priority=10)
d(self.method)
self.assertTrue(self.method._is_clean_step)
self.assertEqual(10, self.method._clean_step_priority)
self.assertFalse(self.method._clean_step_abortable)
self.assertIsNone(self.method._clean_step_argsinfo)
def test_clean_step_all_args(self):
argsinfo = {'arg1': {'description': 'desc1',
'required': True}}
d = driver_base.clean_step(priority=0, abortable=True,
argsinfo=argsinfo)
d(self.method)
self.assertTrue(self.method._is_clean_step)
self.assertEqual(0, self.method._clean_step_priority)
self.assertTrue(self.method._clean_step_abortable)
self.assertEqual(argsinfo, self.method._clean_step_argsinfo)
def test_clean_step_bad_priority(self):
d = driver_base.clean_step(priority='hi')
self.assertRaisesRegexp(exception.InvalidParameterValue, 'priority',
d, self.method)
self.assertTrue(self.method._is_clean_step)
self.assertFalse(hasattr(self.method, '_clean_step_priority'))
self.assertFalse(hasattr(self.method, '_clean_step_abortable'))
self.assertFalse(hasattr(self.method, '_clean_step_argsinfo'))
def test_clean_step_bad_abortable(self):
d = driver_base.clean_step(priority=0, abortable='blue')
self.assertRaisesRegexp(exception.InvalidParameterValue, 'abortable',
d, self.method)
self.assertTrue(self.method._is_clean_step)
self.assertEqual(0, self.method._clean_step_priority)
self.assertFalse(hasattr(self.method, '_clean_step_abortable'))
self.assertFalse(hasattr(self.method, '_clean_step_argsinfo'))
@mock.patch.object(driver_base, '_validate_argsinfo', spec_set=True,
autospec=True)
def test_clean_step_bad_argsinfo(self, mock_valid):
mock_valid.side_effect = exception.InvalidParameterValue('bad')
d = driver_base.clean_step(priority=0, argsinfo=100)
self.assertRaises(exception.InvalidParameterValue, d, self.method)
self.assertTrue(self.method._is_clean_step)
self.assertEqual(0, self.method._clean_step_priority)
self.assertFalse(self.method._clean_step_abortable)
self.assertFalse(hasattr(self.method, '_clean_step_argsinfo'))
class CleanStepTestCase(base.TestCase):
def test_get_and_execute_clean_steps(self):
# Create a fake Driver class, create some clean steps, make sure