openstack/ironic
Code Issues Proposed changes

Follow-up: Apply Inspection Rules

Browse Source 
Migrating Inspector to Ironic

Change-Id: I6ff32eb2739d68d1e922b0b64e7c104f7b3a27a6
This commit is contained in:
cid
2025-03-10 20:03:09 +01:00
parent 3603e60c32
commit de5988af29
13 changed files with 1317 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ironic/common/exception.py
+21
View File
@@ -1095,3 +1095,24 @@ class InspectionRuleAlreadyExists(Conflict):
class InspectionRuleNotFound(NotFound):
"""The requested rule was not found."""
_msg_fmt = _("Rule %(rule)s could not be found.")
class InspectionRuleValidationFailure(IronicException):
"""Inspection rule validation fails during creation or execution."""
_msg_fmt = _("Inspection rule validation failed. Reason: %(reason)s")
class InspectionRuleExecutionFailure(HardwareInspectionFailure):
"""Raised when an inspection rule fails during execution."""
_msg_fmt = _("Inspection rule execution failed. Reason: %(reason)s")
class RuleActionExecutionFailure(InspectionRuleExecutionFailure):
"""Raised when an inspection rule action fails during execution."""
_msg_fmt = _("Inspection rule action execution failed. "
"Reason: %(reason)s")
class RuleConditionCheckFailure(InspectionRuleExecutionFailure):
"""Raised when an inspection rule condition fails during execution."""
_msg_fmt = _("Inspection rule condition check failed. Reason: %(reason)s")
ironic/common/inspection_rules/actions.py
+81 -160
View File
@@ -1,7 +1,3 @@
# Copyright 2013 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@@ -21,6 +17,8 @@ from oslo_log import log
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.inspection_rules import base
from ironic.common.inspection_rules import utils
from ironic.drivers import utils as driver_utils
from ironic import objects
@@ -62,9 +60,6 @@ def update_nested_dict(d, key_path, value):
class ActionBase(base.Base, metaclass=abc.ABCMeta):
"""Abstract base class for rule action plugins."""
OPTIONAL_ARGS = set()
"""Set with names of optional parameters."""
FORMATTED_ARGS = []
"""List of params to be formatted with python format."""
@@ -72,7 +67,7 @@ class ActionBase(base.Base, metaclass=abc.ABCMeta):
def __call__(self, task, *args, **kwargs):
"""Run action on successful rule match."""
def _execute_with_loop(self, task, action, inventory, plugin_data):
def execute_with_loop(self, task, action, inventory, plugin_data):
loop_items = action.get('loop', [])
results = []
@@ -80,55 +75,43 @@ class ActionBase(base.Base, metaclass=abc.ABCMeta):
for item in loop_items:
action_copy = action.copy()
action_copy['args'] = item
results.append(self._execute_action(task, action_copy,
inventory, plugin_data))
results.append(self.execute_action(task, action_copy,
inventory, plugin_data))
return results
def _execute_action(self, task, action, inventory, plugin_data):
def execute_action(self, task, action, inventory, plugin_data):
processed_args = self._process_args(task, action, inventory,
plugin_data)
arg_values = [processed_args[arg_name]
for arg_name in self.get_arg_names()]
for optional_arg in self.OPTIONAL_ARGS:
arg_values.append(processed_args.get(optional_arg, False))
return self(task, *arg_values)
return self(task, **processed_args)
class LogAction(ActionBase):
FORMATTED_ARGS = ['msg']
@classmethod
def get_arg_names(cls):
return ['msg']
FORMATTED_ARGS = ['msg']
VALID_LOG_LEVELS = {'debug', 'info', 'warning', 'error', 'critical'}
def __call__(self, task, msg, level='info'):
level = level.lower()
if level not in self.VALID_LOG_LEVELS:
raise exception.InspectionRuleExecutionFailure(
_("Invalid log level: %(level)s. Choose from %(levels)s") % {
'level': level, 'levels': self.VALID_LOG_LEVELS})
getattr(LOG, level)(msg)
class FailAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['msg']
def __call__(self, task, msg):
msg = _('%(msg)s') % {'msg': msg}
raise exception.HardwareInspectionFailure(error=msg)
raise exception.HardwareInspectionFailure(error=str(msg))
class SetAttributeAction(ActionBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['path', 'value']
FORMATTED_ARGS = ['value']
def __call__(self, task, path, value):
try:
attr_path_parts = path.strip('/').split('/')
attr_path_parts = utils.normalize_path(path)
if len(attr_path_parts) == 1:
setattr(task.node, attr_path_parts[0], value)
else:
@@ -140,25 +123,20 @@ class SetAttributeAction(ActionBase):
setattr(task.node, attr_path_parts[0], base_attr)
task.node.save()
except Exception as exc:
msg = ("Failed to set attribute %(path)s "
"with value %(value)s: %(exc)s" %
{'path': path, 'value': value, 'exc': exc})
msg = _("Failed to set attribute %(path)s "
"with value %(value)s: %(exc)s") % {
'path': path, 'value': value, 'exc': exc}
LOG.error(msg)
raise exception.InvalidParameterValue(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
class ExtendAttributeAction(ActionBase):
OPTIONAL_ARGS = {'unique'}
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['path', 'value']
def __call__(self, task, path, value, unique=False):
try:
attr_path_parts = path.strip('/').split('/')
attr_path_parts = utils.normalize_path(path)
if len(attr_path_parts) == 1:
current = getattr(task.node, attr_path_parts[0], [])
else:
@@ -169,7 +147,9 @@ class ExtendAttributeAction(ActionBase):
current = current.setdefault(attr_path_parts[-1], [])
if not isinstance(current, list):
current = []
msg = _("Cannot extend non-list attribute %(path)s with "
"value %(value)s") % {'path': path, 'value': value}
raise exception.RuleActionExecutionFailure(reason=msg)
if not unique or value not in current:
current.append(value)
@@ -179,20 +159,16 @@ class ExtendAttributeAction(ActionBase):
setattr(task.node, attr_path_parts[0], base_attr)
task.node.save()
except Exception as exc:
msg = ("Failed to extend attribute %(path)s: %(exc)s") % {
msg = _("Failed to extend attribute %(path)s: %(exc)s") % {
'path': path, 'exc': exc}
raise exception.InvalidParameterValue(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
class DelAttributeAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['path']
def __call__(self, task, path):
try:
attr_path_parts = path.strip('/').split('/')
attr_path_parts = utils.normalize_path(path)
if len(attr_path_parts) == 1:
delattr(task.node, attr_path_parts[0])
else:
@@ -204,34 +180,26 @@ class DelAttributeAction(ActionBase):
setattr(task.node, attr_path_parts[0], base_attr)
task.node.save()
except Exception as exc:
msg = ("Failed to delete attribute at %(path)s: %(exc)s") % {
msg = _("Failed to delete attribute at %(path)s: %(exc)s") % {
'path': path, 'exc': exc}
raise exception.InvalidParameterValue(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
class AddTraitAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['name']
def __call__(self, task, name):
try:
new_trait = objects.Trait(task.context, node_id=task.node.id,
trait=name)
new_trait.create()
except Exception as exc:
msg = (_("Failed to add new trait %(name)s: %(exc)s") %
{'name': name, 'exc': exc})
raise exception.InvalidParameterValue(msg)
msg = _("Failed to add new trait %(name)s: %(exc)s") % {
'name': name, 'exc': exc}
raise exception.RuleActionExecutionFailure(reason=msg)
class RemoveTraitAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['name']
def __call__(self, task, name):
try:
objects.Trait.destroy(task.context, node_id=task.node.id,
@@ -240,100 +208,65 @@ class RemoveTraitAction(ActionBase):
LOG.warning(_("Failed to remove trait %(name)s: %(exc)s"),
{'name': name, 'exc': exc})
except Exception as exc:
msg = (_("Failed to remove trait %(name)s: %(exc)s") %
{'name': name, 'exc': exc})
raise exception.InvalidParameterValue(msg)
msg = _("Failed to remove trait %(name)s: %(exc)s") % {
'name': name, 'exc': exc}
raise exception.RuleActionExecutionFailure(reason=msg)
class SetCapabilityAction(ActionBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['name', 'value']
FORMATTED_ARGS = ['value']
def __call__(self, task, name, value):
try:
properties = task.node.properties.copy()
capabilities = properties.get('capabilities', '')
caps = dict(cap.split(':', 1)
for cap in capabilities.split(',') if cap)
caps[name] = value
properties['capabilities'] = ','.join('%s:%s' % (k, v)
for k, v in caps.items())
task.node.properties = properties
task.node.save()
driver_utils.add_node_capability(task, name, value)
except Exception as exc:
raise exception.InvalidParameterValue(
"Failed to set capability %(name)s: %(exc)s" %
{'name': name, 'exc': exc})
msg = _("Failed to set capability %(name)s: %(exc)s") % {
'name': name, 'exc': exc}
raise exception.RuleActionExecutionFailure(
reason=msg)
class UnsetCapabilityAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['name']
def __call__(self, task, name):
try:
properties = task.node.properties.copy()
capabilities = properties.get('capabilities', '')
caps = dict(cap.split(':', 1)
for cap in capabilities.split(',') if cap)
caps.pop(name, None)
properties['capabilities'] = ','.join('%s:%s' % (k, v)
for k, v in caps.items())
task.node.properties = properties
task.node.save()
driver_utils.remove_node_capability(task, name)
except Exception as exc:
raise exception.InvalidParameterValue(
"Failed to unset capability %(name)s: %(exc)s" %
{'name': name, 'exc': exc})
msg = _("Failed to unset capability %(name)s: %(exc)s") % {
'name': name, 'exc': exc}
raise exception.RuleActionExecutionFailure(reason=msg)
class SetPluginDataAction(ActionBase):
REQUIRES_PLUGIN_DATA = True
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['path', 'value', 'plugin_data']
def __call__(self, task, path, value, plugin_data):
try:
update_nested_dict(plugin_data, path, value)
return {'plugin_data': plugin_data}
except Exception as exc:
msg = ("Failed to set plugin data at %(path)s: %(exc)s" % {
'path': path, 'exc': exc})
raise exception.InvalidParameterValue(msg)
msg = _("Failed to set plugin data at %(path)s: %(exc)s") % {
'path': path, 'exc': exc}
raise exception.RuleActionExecutionFailure(reason=msg)
class ExtendPluginDataAction(ActionBase):
OPTIONAL_ARGS = {'unique'}
REQUIRES_PLUGIN_DATA = True
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['path', 'value', 'plugin_data']
def __call__(self, task, path, value, plugin_data, unique=False):
try:
current = self._get_nested_value(plugin_data, path)
if current is None:
current = []
update_nested_dict(plugin_data, path, current)
elif not isinstance(current, list):
current = []
update_nested_dict(plugin_data, path, current)
if not unique or value not in current:
current = self._get_nested_value(plugin_data, path) or []
update_nested_dict(plugin_data, path, current)
if not unique or (value not in current):
current.append(value)
return {'plugin_data': plugin_data}
except Exception as exc:
msg = ("Failed to extend plugin data at %(path)s: %(exc)s") % {
msg = _("Failed to extend plugin data at %(path)s: %(exc)s") % {
'path': path, 'exc': exc}
raise exception.InvalidParameterValue(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
@staticmethod
def _get_nested_value(d, key_path, default=None):
@@ -349,19 +282,16 @@ class ExtendPluginDataAction(ActionBase):
class UnsetPluginDataAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['path', 'plugin_data']
REQUIRES_PLUGIN_DATA = True
def __call__(self, task, path, plugin_data):
try:
if not self._unset_nested_dict(plugin_data, path):
LOG.warning("Path %s not found", path)
return {'plugin_data': plugin_data}
except Exception as exc:
msg = ("Failed to unset plugin data at %(path)s: %(exc)s") % {
msg = _("Failed to unset plugin data at %(path)s: %(exc)s") % {
'path': path, 'exc': exc}
raise exception.InvalidParameterValue(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
@staticmethod
def _unset_nested_dict(d, key_path):
@@ -387,18 +317,15 @@ class UnsetPluginDataAction(ActionBase):
class SetPortAttributeAction(ActionBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['port_id', 'path', 'value']
FORMATTED_ARGS = ['value']
def __call__(self, task, port_id, path, value):
port = next((p for p in task.ports if p.uuid == port_id), None)
if not port:
raise exception.PortNotFound(port=port_id)
try:
attr_path_parts = path.strip('/').split('/')
attr_path_parts = utils.normalize_path(path)
if len(attr_path_parts) == 1:
setattr(port, attr_path_parts[0], value)
else:
@@ -410,27 +337,23 @@ class SetPortAttributeAction(ActionBase):
setattr(port, attr_path_parts[0], base_attr)
port.save()
except Exception as exc:
msg = ("Failed to set attribute %(path)s for port "
"%(port_id)s: %(exc)s") % {'path': path,
'port_id': port_id,
'exc': str(exc)}
LOG.warning(msg)
msg = _("Failed to set attribute %(path)s for port "
"%(port_id)s: %(exc)s") % {
'path': path, 'port_id': port_id, 'exc': exc}
LOG.error(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
class ExtendPortAttributeAction(ActionBase):
OPTIONAL_ARGS = {'unique'}
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['port_id', 'path', 'value']
FORMATTED_ARGS = ['value']
def __call__(self, task, port_id, path, value, unique=False):
port = next((p for p in task.ports if p.uuid == port_id), None)
if not port:
raise exception.PortNotFound(port=port_id)
try:
attr_path_parts = path.strip('/').split('/')
attr_path_parts = utils.normalize_path(path)
if len(attr_path_parts) == 1:
current = getattr(port, attr_path_parts[0], [])
else:
@@ -441,7 +364,9 @@ class ExtendPortAttributeAction(ActionBase):
current = current.setdefault(attr_path_parts[-1], [])
if not isinstance(current, list):
current = []
msg = (_("Cannot extend non-list attribute %(path)s with "
" value %(value)s") % {'path': path, 'value': value})
raise exception.RuleActionExecutionFailure(reason=msg)
if not unique or value not in current:
current.append(value)
@@ -451,25 +376,21 @@ class ExtendPortAttributeAction(ActionBase):
setattr(port, attr_path_parts[0], base_attr)
port.save()
except Exception as exc:
msg = ("Failed to extend attribute %(path)s for port "
"%(port_id)s: %(exc)s") % {'path': path,
'port_id': port_id,
'exc': str(exc)}
LOG.warning(msg)
msg = _("Failed to extend attribute %(path)s for port "
"%(port_id)s: %(exc)s") % {
'path': path, 'port_id': port_id, 'exc': exc}
LOG.error(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
class DelPortAttributeAction(ActionBase):
@classmethod
def get_arg_names(cls):
return ['port_id', 'path']
def __call__(self, task, port_id, path):
port = next((p for p in task.ports if p.uuid == port_id), None)
if not port:
raise exception.PortNotFound(port=port_id)
try:
attr_path_parts = path.strip('/').split('/')
attr_path_parts = utils.normalize_path(path)
if len(attr_path_parts) == 1:
delattr(port, attr_path_parts[0])
else:
@@ -482,7 +403,7 @@ class DelPortAttributeAction(ActionBase):
port.save()
except Exception as exc:
msg = ("Failed to delete attribute %(path)s for port "
"%(port_id)s: %(exc)s") % {'path': path,
'port_id': port_id,
'exc': str(exc)}
LOG.warning(msg)
"%(port_id)s: %(exc)s") % {
'path': path, 'port_id': port_id, 'exc': str(exc)}
LOG.error(msg)
raise exception.RuleActionExecutionFailure(reason=msg)
ironic/common/inspection_rules/base.py
+58 -56
View File
@@ -1,6 +1,3 @@
# Copyright 2013 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@@ -13,12 +10,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import abc
import inspect
from oslo_log import log
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common import utils as common_utils
from ironic.common.inspection_rules import utils
import ironic.conf
@@ -29,70 +27,69 @@ SENSITIVE_FIELDS = ['password', 'auth_token', 'bmc_password']
class Base(object):
USES_PLUGIN_DATA = False
REQUIRES_PLUGIN_DATA = False
"""Flag to indicate if this action needs plugin_data as an arg."""
OPTIONAL_ARGS = set()
"""Set with names of optional parameters."""
def _get_validation_signature(self):
"""Get the signature to validate against."""
signature = inspect.signature(self.__call__)
@classmethod
@abc.abstractmethod
def get_arg_names(cls):
"""Return list of argument names in order expected."""
raise NotImplementedError
# Strip off 'task' parameter.
parameters = list(signature.parameters.values())[1:]
def _normalize_list_args(self, *args, **kwargs):
"""Convert list arguments into dictionary format.
required_args = [p.name for p in parameters
if p.default is inspect.Parameter.empty]
optional_args = [p.name for p in parameters
if p.default is not inspect.Parameter.empty]
return required_args, optional_args
"""
op_name = kwargs['op']
arg_list = kwargs['args']
if not isinstance(arg_list, list):
if isinstance(arg_list, dict) and 'plugin-data' in op_name:
arg_list['plugin_data'] = {}
return arg_list
def _normalize_list_args(self, required_args, optional_args, op_args):
"""Convert list arguments into dictionary format."""
if not isinstance(op_args, list):
# Initialize required context fields if needed
if isinstance(op_args, dict) and self.REQUIRES_PLUGIN_DATA:
op_args['plugin_data'] = {}
return op_args
# plugin_data is a required argument during validation but since
# it comes from the inspection data and added later, we need to
# make sure validation does not fail for that sake.
if 'plugin-data' in op_name:
arg_list.append('{}')
# Initialize required context fields if needed
if self.REQUIRES_PLUGIN_DATA:
op_args.append({})
arg_names = set(self.__class__.get_arg_names())
if len(arg_list) < len(arg_names):
missing = arg_names[len(arg_list):]
if len(op_args) < len(required_args):
missing = [p for p in required_args[len(op_args):]]
msg = (_("Not enough arguments provided. Missing: %s"),
", ".join(missing))
LOG.error(msg)
raise ValueError(msg)
raise exception.InspectionRuleValidationFailure(msg)
arg_list = {name: arg_list[i] for i, name in enumerate(arg_names)}
normalized_args = {name: op_args[i]
for i, name in enumerate(required_args)}
# Add optional args if they exist in the input
start_idx = len(arg_names)
for i, opt_arg in enumerate(self.OPTIONAL_ARGS):
if start_idx + i < len(arg_list):
arg_list[opt_arg] = arg_list[start_idx + i]
normalized_args.update(
zip(optional_args, op_args[len(required_args):])
)
return arg_list
return normalized_args
def validate(self, *args, **kwargs):
def validate(self, op_args):
"""Validate args passed during creation.
Default implementation checks for presence of required fields.
:param args: args as a dictionary
:param kwargs: used for extensibility without breaking existing plugins
:raises: ValueError on validation failure
:param op_args: Operator args as a dictionary
:raises: InspectionRuleValidationFailure on validation failure
"""
required_args = set(self.__class__.get_arg_names())
required_args, optional_args = self._get_validation_signature()
normalized_args = self._normalize_list_args(
args=kwargs.get('args', {}), op=kwargs['op'])
required_args=required_args, optional_args=optional_args,
op_args=op_args)
# If after normalization attempt, we still do not have a dictionary,
# then it was never a list, so, not a supported type.
if isinstance(normalized_args, dict):
provided = set(normalized_args.keys())
missing = required_args - provided
unexpected = provided - (required_args | self.OPTIONAL_ARGS)
provided = set(normalized_args)
missing = set(required_args) - provided
unexpected = provided - (set(required_args) | set(optional_args))
msg = []
if missing:
@@ -102,11 +99,12 @@ class Base(object):
msg.append(_('unexpected argument(s): %s')
% ', '.join(unexpected))
if msg:
raise ValueError('; '.join(msg))
raise exception.InspectionRuleValidationFailure(
'; '.join(msg))
else:
raise ValueError(_("args must be either a list or dictionary"))
raise exception.InspectionRuleValidationFailure(
_("args must be either a list or dictionary"))
@staticmethod
def interpolate_variables(value, node, inventory, plugin_data):
if isinstance(value, str):
try:
@@ -135,21 +133,25 @@ class Base(object):
op = operation.get('op')
if not op:
raise ValueError("Operation must contain 'op' key")
raise exception.InspectionRuleExecutionFailure(
_("Operation must contain 'op' key"))
op, invtd = common_utils.parse_inverted_operator(op)
dict_args = self._normalize_list_args(args=operation.get('args', {}),
op=op)
required_args, optional_args = self._get_validation_signature()
op, invtd = utils.parse_inverted_operator(op)
dict_args = self._normalize_list_args(
required_args=required_args, optional_args=optional_args,
op_args=operation.get('args', {}))
# plugin-data becomes available during inspection,
# we need to populate with the actual value.
if 'plugin_data' in dict_args or 'plugin-data' in op:
if self.REQUIRES_PLUGIN_DATA:
dict_args['plugin_data'] = plugin_data
node = task.node
formatted_args = getattr(self, 'FORMATTED_ARGS', [])
return {
k: (self.interpolate_variables(v, node, inventory, plugin_data)
k: (Base.interpolate_variables(v, node, inventory, plugin_data)
if k in formatted_args else v)
for k, v in dict_args.items()
}
ironic/common/inspection_rules/engine.py
+36 -61
View File
@@ -1,6 +1,3 @@
# Copyright 2013 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@@ -13,7 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log
import yaml
@@ -21,8 +17,8 @@ from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.inspection_rules import actions
from ironic.common.inspection_rules import operators
from ironic.common.inspection_rules import utils
from ironic.common.inspection_rules import validation
from ironic.common import utils as common_utils
from ironic.conf import CONF
from ironic import objects
@@ -56,7 +52,7 @@ def get_built_in_rules():
'conditions': rule_data.get('conditions', []),
'built_in': True
}
validation.validate_inspection_rule(rule)
validation.validate_rule(rule)
built_in_rules.append(rule)
except Exception as e:
LOG.error(_("Error parsing built-in rule: %s"), e)
@@ -77,24 +73,13 @@ def get_built_in_rules():
return built_in_rules
def _mask_sensitive_data(data):
"""Recursively mask sensitive fields in data."""
if isinstance(data, dict):
return {key: (_mask_sensitive_data(value)
if key not in SENSITIVE_FIELDS else '***')
for key, value in data.items()}
elif isinstance(data, list):
return [_mask_sensitive_data(item) for item in data]
return data
def check_conditions(task, rule, inventory, plugin_data):
try:
if not rule.get('conditions', None):
return True
for condition in rule['conditions']:
op, invtd = common_utils.parse_inverted_operator(
op, invtd = utils.parse_inverted_operator(
condition['op'])
if op not in operators.OPERATORS:
@@ -104,14 +89,13 @@ def check_conditions(task, rule, inventory, plugin_data):
'op': op, 'supported_ops': supported_ops})
raise ValueError(msg)
result = False
plugin = operators.get_operator(op)
if 'loop' in condition:
result = plugin()._check_with_loop(task, condition, inventory,
plugin_data)
result = plugin().check_with_loop(task, condition, inventory,
plugin_data)
else:
result = plugin()._check_condition(task, condition, inventory,
plugin_data)
result = plugin().check_condition(task, condition, inventory,
plugin_data)
if not result:
LOG.debug("Skipping rule %(rule)s on node %(node)s: "
"condition check '%(op)s': '%(args)s' failed ",
@@ -128,7 +112,6 @@ def check_conditions(task, rule, inventory, plugin_data):
def apply_actions(task, rule, inventory, plugin_data):
result = {'plugin_data': plugin_data}
for action in rule['actions']:
try:
op = action['op']
@@ -141,15 +124,11 @@ def apply_actions(task, rule, inventory, plugin_data):
plugin = actions.get_action(op)
if 'loop' in action:
action_result = plugin()._execute_with_loop(
task, action, inventory, result['plugin_data'])
plugin().execute_with_loop(task, action, inventory,
plugin_data)
else:
action_result = plugin()._execute_action(
task, action, inventory, result['plugin_data'])
if action_result is not None and isinstance(action_result, dict):
result['plugin_data'] = action_result.get(
'plugin_data', result['plugin_data'])
plugin().execute_action(task, action, inventory,
plugin_data)
except exception.IronicException as err:
LOG.error("Error applying action on node %(node)s: %(err)s.",
{'node': task.node.uuid, 'err': err})
@@ -159,7 +138,6 @@ def apply_actions(task, rule, inventory, plugin_data):
"%(node)s: %(err)s.", {'node': task.node.uuid,
'err': err})
raise
return result
def apply_rules(task, inventory, plugin_data, inspection_phase):
@@ -180,47 +158,45 @@ def apply_rules(task, inventory, plugin_data, inspection_phase):
'node': node.uuid})
return
mask_secrets = CONF.inspection_rules.mask_secrets
if mask_secrets == 'always':
inventory = _mask_sensitive_data(inventory)
plugin_data = _mask_sensitive_data(plugin_data)
elif mask_secrets == 'sensitive':
# Mask secrets unless the rule is marked as sensitive
for rule in rules:
if not rule.get('sensitive', False):
inventory = _mask_sensitive_data(inventory)
plugin_data = _mask_sensitive_data(plugin_data)
break
rules.sort(key=lambda rule: rule['priority'], reverse=True)
rules.sort(key=lambda rule: rule.get('priority', 0), reverse=True)
LOG.debug("Applying %(count)d inspection rules to node %(node)s",
{'count': len(rules), 'node': node.uuid})
result = {'plugin_data': plugin_data}
mask_secrets = CONF.inspection_rules.mask_secrets
for rule in rules:
try:
if not check_conditions(task, rule, inventory, plugin_data):
should_mask = False
is_sensitive_rule = rule.get('sensitive', False)
if (mask_secrets == 'always'
or mask_secrets == 'sensitive' and not is_sensitive_rule):
should_mask = True
masked_inventory = utils.ShallowMaskDict(
inventory, sensitive_fields=SENSITIVE_FIELDS,
mask_enabled=should_mask)
masked_plugin_data = utils.ShallowMaskDict(
plugin_data, sensitive_fields=SENSITIVE_FIELDS,
mask_enabled=should_mask)
if not check_conditions(task, rule, masked_inventory,
masked_plugin_data):
continue
LOG.info("Applying actions for rule %(rule)s to node %(node)s",
{'rule': rule['uuid'], 'node': node.uuid})
rule_result = apply_actions(task, rule, inventory, plugin_data)
if rule_result and 'plugin_data' in rule_result:
result['plugin_data'] = rule_result['plugin_data']
apply_actions(task, rule, masked_inventory, masked_plugin_data)
except exception.HardwareInspectionFailure:
raise
except exception.IronicException as e:
if rule['sensitive']:
LOG.error("Error applying sensitive rule %(rule)s to node "
"%(node)s", {'rule': rule['uuid'],
'node': node.uuid})
else:
LOG.error("Error applying rule %(rule)s to node "
"%(node)s: %(error)s", {'rule': rule['uuid'],
'node': node.uuid,
'error': e})
LOG.error(_("Error applying rule %(rule)s to node "
"%(node)s: %(error)s"), {'rule': rule['uuid'],
'node': node.uuid,
'error': e})
raise
except Exception as e:
msg = ("Failed to apply rule %(rule)s to node %(node)s: "
@@ -232,4 +208,3 @@ def apply_rules(task, inventory, plugin_data, inspection_phase):
raise exception.IronicException(msg)
LOG.info("Finished applying inspection rules to node %s", node.uuid)
return result
ironic/common/inspection_rules/operators.py
+47 -70
View File
@@ -1,6 +1,3 @@
# Copyright 2013 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@@ -20,9 +17,10 @@ import re
import netaddr
from oslo_log import log
from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.inspection_rules import base
from ironic.common import utils as common_utils
from ironic.common.inspection_rules import utils
LOG = log.getLogger(__name__)
@@ -59,14 +57,11 @@ def coerce(value, expected):
class OperatorBase(base.Base, metaclass=abc.ABCMeta):
"""Abstract base class for rule condition plugins."""
OPTIONAL_ARGS = set()
"""Set with names of optional parameters."""
@abc.abstractmethod
def check(self, *args, **kwargs):
"""Check if condition holds for a given field."""
def __call__(self, task, *args, **kwargs):
"""Checks if condition holds for a given field."""
def _check_with_loop(self, task, condition, inventory, plugin_data):
def check_with_loop(self, task, condition, inventory, plugin_data):
loop_items = condition.get('loop', [])
multiple = condition.get('multiple', 'any')
results = []
@@ -75,8 +70,8 @@ class OperatorBase(base.Base, metaclass=abc.ABCMeta):
for item in loop_items:
condition_copy = condition.copy()
condition_copy['args'] = item
result = self._check_condition(task, condition_copy,
inventory, plugin_data)
result = self.check_condition(task, condition_copy,
inventory, plugin_data)
results.append(result)
if multiple == 'first' and result:
@@ -89,9 +84,9 @@ class OperatorBase(base.Base, metaclass=abc.ABCMeta):
elif multiple == 'all':
return all(results)
return results[0] if results else False
return self._check_condition(task, condition, inventory, plugin_data)
return self.check_condition(task, condition, inventory, plugin_data)
def _check_condition(self, task, condition, inventory, plugin_data):
def check_condition(self, task, condition, inventory, plugin_data):
"""Process condition arguments and apply the check logic.
:param task: TaskManger instance
@@ -99,37 +94,41 @@ class OperatorBase(base.Base, metaclass=abc.ABCMeta):
:param args: parameters as a dictionary, changing it here will change
what will be stored in database
:param kwargs: used for extensibility without breaking existing plugins
:raises ValueError: on unacceptable field value
:raises InspectionRuleExecutionFailure: on unacceptable field value
:returns: True if check succeeded, otherwise False
"""
op, is_inverted = common_utils.parse_inverted_operator(
op, is_inverted = utils.parse_inverted_operator(
condition['op'])
processed_args = self._process_args(task, condition, inventory,
plugin_data)
arg_values = [processed_args[arg_name]
for arg_name in self.get_arg_names()]
for optional_arg in self.OPTIONAL_ARGS:
arg_values.append(processed_args.get(optional_arg, False))
result = self.check(*arg_values)
result = self(task, **processed_args)
return not result if is_inverted else result
class SimpleOperator(OperatorBase):
op = None
OPTIONAL_ARGS = {'force_strings'}
@classmethod
def get_arg_names(cls):
return ['values']
def __call__(self, task, values, force_strings=False):
if not isinstance(values, list):
msg = _("Failed to check condition: '%(op)s' on values: "
"%(values)s: Expected list for 'values', got: "
"%(invalid_type)s") % {
'op': self.op.__name__, 'values': values,
"invalid_type": type(values).__name__}
LOG.error(msg)
raise exception.RuleConditionCheckFailure(reason=msg)
if len(values) < 2:
return True
def check(self, values, force_strings=False):
if force_strings:
values = [coerce(value, str) for value in values]
return self.op(values)
return all(self.op(values[i], values[i + 1])
for i in range(len(values) - 1))
class EqOperator(SimpleOperator):
@@ -145,42 +144,31 @@ class GtOperator(SimpleOperator):
class EmptyOperator(OperatorBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['value']
def check(self, value):
def __call__(self, task, value):
return str(value) in ("", 'None', '[]', '{}')
class NetOperator(OperatorBase):
FORMATTED_ARGS = ['address', 'subnet']
@classmethod
def get_arg_names(cls):
return ['address', 'subnet']
def validate(self, address, subnet):
def __call__(self, task, address, subnet):
try:
netaddr.IPNetwork(subnet)
network = netaddr.IPNetwork(subnet)
except netaddr.AddrFormatError as exc:
LOG.error(_('invalid value: %s'), exc)
def check(self, address, subnet):
network = netaddr.IPNetwork(subnet)
raise exception.InspectionRuleExecutionFailure(
_('invalid value: %s') % exc)
return netaddr.IPAddress(address) in network
class IsTrueOperator(OperatorBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['value']
def check(self, value):
def __call__(self, task, value):
if isinstance(value, bool):
return value
if isinstance(value, (int, float)):
@@ -191,13 +179,10 @@ class IsTrueOperator(OperatorBase):
class IsFalseOperator(OperatorBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['value']
def check(self, value):
def __call__(self, task, value):
if isinstance(value, bool):
return not value
if isinstance(value, (int, float)):
@@ -208,44 +193,36 @@ class IsFalseOperator(OperatorBase):
class IsNoneOperator(OperatorBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['value']
def check(self, value):
def __call__(self, task, value):
return str(value) == 'None'
class OneOfOperator(OperatorBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['value', 'values']
def check(self, value, values=[]):
def __call__(self, task, value, values=[]):
return value in values
class ReOperator(OperatorBase):
FORMATTED_ARGS = ['value']
@classmethod
def get_arg_names(cls):
return ['value', 'regex']
FORMATTED_ARGS = ['value']
def validate_regex(self, regex):
try:
re.compile(regex)
except re.error as exc:
raise ValueError(_('invalid regular expression: %s') % exc)
raise exception.InspectionRuleExecutionFailure(
_('invalid regular expression: %s') % exc)
class MatchesOperator(ReOperator):
def check(self, value, regex):
def __call__(self, task, value, regex):
self.validate_regex(regex)
if regex[-1] != '$':
regex += '$'
@@ -254,6 +231,6 @@ class MatchesOperator(ReOperator):
class ContainsOperator(ReOperator):
def check(self, value, regex):
def __call__(self, task, value, regex):
self.validate_regex(regex)
return re.search(regex, str(value)) is not None
ironic/common/inspection_rules/utils.py
+181
View File
@@ -0,0 +1,181 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Utilities and helper functions for rules."""
from collections import abc
from ironic.common.i18n import _
class ShallowMaskList(abc.MutableSequence):
"""A proxy list to maintain original list and applies masking on the fly.
This class implements the MutableSequence ABC to provide a complete
list-like interface while handling sensitive data masking consistently
with ShallowMaskDict.
"""
def __init__(self, original_list, sensitive_fields=None,
mask_enabled=True):
self._original = original_list
self._sensitive_fields = sensitive_fields or []
self._mask_enabled = mask_enabled
def _mask_value(self, value):
"""Apply masking to value on demand."""
if isinstance(value, dict):
return ShallowMaskDict(
value,
sensitive_fields=self._sensitive_fields,
mask_enabled=self._mask_enabled
)
elif isinstance(value, list):
return ShallowMaskList(
value,
sensitive_fields=self._sensitive_fields,
mask_enabled=self._mask_enabled
)
return value
def __getitem__(self, index):
value = self._original[index]
return self._mask_value(value)
def __setitem__(self, index, value):
self._original[index] = value
def __delitem__(self, index):
del self._original[index]
def __iter__(self):
for item in self._original:
yield self._mask_value(item)
def __len__(self):
return len(self._original)
def insert(self, index, value):
self._original.insert(index, value)
def copy(self):
return ShallowMaskList(
self._original.copy(),
sensitive_fields=self._sensitive_fields,
mask_enabled=self._mask_enabled
)
def set_mask_enabled(self, mask_enabled):
self._mask_enabled = mask_enabled
def __repr__(self):
items = [repr(self._mask_value(item)) for item in self._original]
return "[%s]" % ", ".join(items)
class ShallowMaskDict(abc.MutableMapping):
"""Dictionary wrapper to mask sensitive fields on the fly.
This class implements the MutableMapping ABC to provide a complete
dict-like interface while masking sensitive fields when accessed.
"""
def __init__(self, data, sensitive_fields=None, mask_enabled=True):
self._data = data
self._sensitive_fields = sensitive_fields or []
self._mask_enabled = mask_enabled
def _mask_value(self, key, value):
if self._mask_enabled and key in self._sensitive_fields:
return '***'
if isinstance(value, dict):
return ShallowMaskDict(
value,
sensitive_fields=self._sensitive_fields,
mask_enabled=self._mask_enabled
)
elif isinstance(value, list):
return ShallowMaskList(
value,
sensitive_fields=self._sensitive_fields,
mask_enabled=self._mask_enabled
)
return value
def __getitem__(self, key):
value = self._data[key]
return self._mask_value(key, value)
def __setitem__(self, key, value):
self._data[key] = value
def __delitem__(self, key):
del self._data[key]
def __iter__(self):
return iter(self._data)
def __len__(self):
return len(self._data)
def copy(self):
return ShallowMaskDict(
self._data.copy(),
sensitive_fields=self._sensitive_fields,
mask_enabled=self._mask_enabled,
)
def set_mask_enabled(self, mask_enabled):
self._mask_enabled = mask_enabled
def __repr__(self):
items = ["%s: %s" % (repr(k), repr(self._mask_value(k, v)))
for k, v in self._data.items()]
return "{%s}" % ", ".join(items)
def parse_inverted_operator(op):
"""Handle inverted operators.
Parses a logical condition operator to determine if it has been negated
using a single leading exclamation mark ('!'). Ensures only one
exclamation mark is allowed.
Example Usage:
parse_inverted_operator("!eq") # Returns ("eq", True)
parse_inverted_operator(" eq ") # Returns ("eq", False)
parse_inverted_operator("!!eq") # Raises ValueError
raises: ValueError: If multiple exclamation marks are present
returns: A tuple containing the cleaned operator and a
boolean indicating whether negation was applied.
"""
op = op.strip()
if op.count('!') > 1:
msg = _("Multiple exclamation marks are not allowed. "
"To apply the invert of an operation, simply add an "
"exclamation mark (with an optional space) before "
"the operator, e.g. eq - !eq.")
raise ValueError(msg)
is_inverted = op.startswith('!')
op = op.lstrip('!').strip()
return op, is_inverted
def normalize_path(path):
"""Convert a path (dot or slash notation) to a list of path parts"""
if '/' in path:
parts = path.strip('/').split('/')
else:
parts = path.split('.')
return parts
ironic/common/inspection_rules/validation.py
+5 -8
View File
@@ -1,6 +1,3 @@
# Copyright 2013 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
@@ -22,7 +19,7 @@ from ironic.common import exception
from ironic.common.i18n import _
from ironic.common.inspection_rules import actions
from ironic.common.inspection_rules import operators
from ironic.common import utils as common_utils
from ironic.common.inspection_rules import utils
_CONDITIONS_SCHEMA = None
@@ -125,7 +122,7 @@ VALIDATOR = args.and_valid(
)
def validate_inspection_rule(rule):
def validate_rule(rule):
"""Validate an inspection rule using the JSON schema.
:param rule: The inspection rule to validate.
@@ -158,14 +155,14 @@ def validate_inspection_rule(rule):
# Additional plugin-specific validation
for condition in rule.get('conditions', []):
op, invtd = common_utils.parse_inverted_operator(
op, invtd = utils.parse_inverted_operator(
condition['op'])
plugin = operators.get_operator(op)
if not plugin or not callable(plugin):
errors.append(
_('Unsupported condition operator: %s') % op)
try:
plugin().validate(**condition)
plugin().validate(condition.get('args', {}))
except ValueError as exc:
errors.append(_('Invalid parameters for condition operator '
'%(op)s: %(error)s') % {'op': op,
@@ -176,7 +173,7 @@ def validate_inspection_rule(rule):
if not plugin or not callable(plugin):
errors.append(_('Unsupported action operator: %s') % action['op'])
try:
plugin().validate(**action)
plugin().validate(action.get('args', {}))
except ValueError as exc:
errors.append(_('Invalid parameters for action operator %(op)s: '
'%(error)s') % {'op': action['op'], 'error': exc})
ironic/common/utils.py
-15
View File
@@ -1147,18 +1147,3 @@ def get_route_source(dest, ignore_link_local=True):
except (IndexError, ValueError):
LOG.debug('No route to host %(dest)s, route record: %(rec)s',
{'dest': dest, 'rec': out})
def parse_inverted_operator(op):
"""Handle inverted operators."""
op = op.strip()
if op.count('!') > 1:
msg = _("Multiple exclamation marks are not allowed. "
"To apply the invert of an operation, simply put an "
"exclamation mark (with an optional space) before "
"the op, e.g. eq - !eq.")
raise ValueError(msg)
is_inverted = op.startswith('!')
op = op.lstrip('!').strip()
return op, is_inverted
ironic/conductor/inspection.py
+1 -3
View File
@@ -132,9 +132,7 @@ def continue_inspection(task, inventory, plugin_data):
'asynchronously for node %s', node.uuid)
return
result = engine.apply_rules(task, inventory, plugin_data, 'main')
if result and 'plugin_data' in result:
plugin_data = result['plugin_data']
engine.apply_rules(task, inventory, plugin_data, 'main')
# NOTE(dtantsur): logs can be huge and are stored separately
plugin_data.pop('logs', None)
ironic/drivers/utils.py
+21
View File
@@ -185,6 +185,27 @@ def add_node_capability(task, capability, value):
node.save()
def remove_node_capability(task, name):
"""Remove 'capability' from node's 'capabilities' property.
If 'capability' is empty, do nothing.
:param task: Task object.
:param capability: Capability key.
"""
node = task.node
properties = node.properties
capabilities = properties.get('capabilities', '').split(',')
if not capabilities:
return
updated_capabilities = [cap for cap in capabilities
if not cap.startswith("%s:" % name)]
properties['capabilities'] = ','.join(updated_capabilities or [])
node.properties = properties
node.save()
def ensure_next_boot_device(task, driver_info):
"""Ensure boot from correct device if persistent is True
ironic/tests/unit/common/test_inspection_rule.py
+784
View File
@@ -0,0 +1,784 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from unittest import mock
from oslo_utils import uuidutils
from ironic.common import exception
from ironic.common import inspection_rules
from ironic.common.inspection_rules import engine
from ironic.common.inspection_rules import utils
from ironic.conductor import task_manager
from ironic.tests.unit.db import base as db_base
from ironic.tests.unit.objects import utils as obj_utils
class TestApplyRules(db_base.DbTestCase):
def setUp(self):
super(TestApplyRules, self).setUp()
self.node = obj_utils.create_test_node(self.context,
driver='fake-hardware')
self.sensitive_fields = ['password', 'auth_token', 'bmc_password']
self.test_data = {
'username': 'testuser',
'password': 'secret123',
'nested': {
'normal': 'value',
'password': 'nested_secret'
},
'list_data': [
{'name': 'item1', 'password': 'item1_secret'},
{'name': 'item2', 'normal': 'value2'}
],
'auth_token': 'abc123token'
}
self.inventory = {
'cpu': {'count': 4, 'architecture': 'x86_64'},
'memory': {'total': 8192, 'physical_mb': 8192},
'interfaces': [
{'name': 'eth0', 'mac_address': '2a:03:9c:53:4e:46'},
{'name': 'eth1', 'mac_address': 'a2:67:c1:b8:c1:bd'}
],
'disks': [
{'name': '/dev/sda', 'size': 1000000, 'model': 'test-disk-1'},
{'name': '/dev/sdb', 'size': 2000000, 'model': 'test-disk-2'}
],
'bmc_address': '192.168.1.100',
'bmc_password': 'secret'
}
self.plugin_data = {"plugin": "data", "logs": "test logs",
"password": "plugin_secret"}
self.rule1 = obj_utils.create_test_inspection_rule(self.context)
self.rule2 = obj_utils.create_test_inspection_rule(self.context)
self.sensitive_rule = obj_utils.create_test_inspection_rule(
self.context, sensitive=True)
def test_set_mask_enabled(self):
"""Test that set_mask_enabled properly toggles masking."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
self.assertEqual('***', masked_dict['password'])
masked_dict.set_mask_enabled(False)
self.assertEqual('secret123', masked_dict['password'])
masked_dict.set_mask_enabled(True)
self.assertEqual('***', masked_dict['password'])
def test_getitem_masked(self):
"""Test that __getitem__ masks sensitive fields."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
self.assertEqual('***', masked_dict['password'])
self.assertEqual('***', masked_dict['auth_token'])
self.assertEqual('testuser', masked_dict['username'])
def test_getitem_not_masked(self):
"""Test that __getitem__ doesn't mask when disabled."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=False)
self.assertEqual('secret123', masked_dict['password'])
self.assertEqual('abc123token', masked_dict['auth_token'])
def test_nested_dict_masking(self):
"""Test that nested dictionaries are properly masked."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
nested = masked_dict['nested']
self.assertIsInstance(nested, utils.ShallowMaskDict)
self.assertEqual('***', nested['password'])
self.assertEqual('value', nested['normal'])
def test_list_masking(self):
"""Test that lists containing dictionaries are properly masked."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
list_data = masked_dict['list_data']
self.assertEqual('***', list_data[0]['password'])
self.assertEqual('item1', list_data[0]['name'])
self.assertEqual('value2', list_data[1]['normal'])
def test_items_masked(self):
"""Test that items() method returns masked values."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
items = dict(masked_dict.items())
self.assertEqual('***', items['password'])
self.assertEqual('***', items['auth_token'])
self.assertEqual('***', items['nested']['password'])
def test_values_masked(self):
"""Test that values() method masks sensitive values."""
test_data = {'username': 'user', 'password': 'secret'}
masked_dict = utils.ShallowMaskDict(
test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
values = list(masked_dict.values())
self.assertIn('user', values)
self.assertIn('***', values)
self.assertNotIn('secret', values)
def test_get_method_masked(self):
"""Test that the get() method properly masks sensitive fields."""
masked_dict = utils.ShallowMaskDict(
self.test_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
self.assertEqual('***', masked_dict.get('password'))
self.assertEqual('testuser', masked_dict.get('username'))
# Non-existent field should return default
self.assertIsNone(masked_dict.get('nonexistent'))
self.assertEqual('default', masked_dict.get('nonexistent', 'default'))
def test_modifying_dict(self):
"""Test that modifications affect the original data."""
original_data = {'username': 'user', 'data': [1, 2, 3]}
masked_dict = utils.ShallowMaskDict(
original_data, sensitive_fields=self.sensitive_fields,
mask_enabled=True)
masked_dict['new_key'] = 'new_value'
masked_dict['data'].append(4)
self.assertEqual('new_value', original_data['new_key'])
self.assertEqual([1, 2, 3, 4], original_data['data'])
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_no_rules(self, mock_list, mock_apply_actions,
mock_check_conditions, mock_get_built_in):
mock_list.return_value = []
mock_get_built_in.return_value = []
with task_manager.acquire(self.context, self.node.uuid) as task:
result = engine.apply_rules(task, self.inventory,
self.plugin_data, 'main')
mock_list.assert_called_once_with(
context=self.context,
filters={'phase': 'main'})
mock_get_built_in.assert_called_once()
mock_check_conditions.assert_not_called()
mock_apply_actions.assert_not_called()
self.assertIsNone(result)
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_success(self, mock_list, mock_apply_actions,
mock_check_conditions, mock_get_built_in):
rule1 = {'uuid': 'rule-1', 'priority': 100, 'conditions': [],
'actions': [{'op': 'set-attribute',
'args': {'path': 'a', 'value': 'b'}}]}
rule2 = {'uuid': 'rule-2', 'priority': 50, 'conditions': [],
'actions': [
{'op': 'set-capability',
'args': {'name': 'boot_mode', 'value': 'uefi'}}]}
mock_list.return_value = [rule1]
mock_get_built_in.return_value = [rule2]
mock_check_conditions.return_value = True
with task_manager.acquire(self.context, self.node.uuid) as task:
engine.apply_rules(task, self.inventory, self.plugin_data, 'main')
mock_get_built_in.assert_called_once()
mock_list.assert_called_once_with(
context=self.context,
filters={'phase': 'main'})
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_condition_false(self, mock_list, mock_apply_actions,
mock_check_conditions,
mock_get_built_in):
"""Test that rules are skipped when conditions don't match."""
mock_list.return_value = [self.rule1]
mock_get_built_in.return_value = [self.rule2]
mock_check_conditions.side_effect = [False, True]
with task_manager.acquire(self.context, self.node.uuid) as task:
engine.apply_rules(task, self.inventory, self.plugin_data, 'main')
self.assertEqual(2, mock_check_conditions.call_count)
@mock.patch.object(engine, 'LOG', autospec=True)
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_ironic_exception(self, mock_list,
mock_apply_actions,
mock_check_conditions,
mock_get_built_in,
mock_log):
"""Test that IronicException is re-raised."""
mock_list.return_value = [self.rule1, self.rule2]
mock_get_built_in.return_value = []
mock_check_conditions.return_value = True
mock_apply_actions.side_effect = [
exception.IronicException("Expected error"),
{'plugin_data': {'updated': 'data'}}
]
with task_manager.acquire(self.context, self.node.uuid) as task:
self.assertRaises(exception.IronicException,
engine.apply_rules, task, self.inventory,
self.plugin_data, 'main')
mock_log.error.assert_called_once()
self.assertEqual(1, mock_apply_actions.call_count)
@mock.patch.object(utils, 'ShallowMaskDict', autospec=True)
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_with_always_mask(self, mock_list, mock_apply_actions,
mock_check_conditions,
mock_get_built_in,
mock_masked_dict):
"""Test apply_rules with mask_secrets='always'."""
self.config(mask_secrets='always', group='inspection_rules')
mock_list.return_value = [self.rule1]
mock_get_built_in.return_value = [self.rule2]
mock_check_conditions.return_value = True
masked_inventory = mock.MagicMock()
masked_plugin_data = mock.MagicMock()
mock_masked_dict.side_effect = [masked_inventory, masked_plugin_data,
mock.MagicMock(), mock.MagicMock()]
with task_manager.acquire(self.context, self.node.uuid) as task:
engine.apply_rules(task, self.inventory, self.plugin_data, 'main')
mock_masked_dict.assert_has_calls([
mock.call(self.inventory,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=True),
mock.call(self.plugin_data,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=True),
mock.call(self.inventory,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=True),
mock.call(self.plugin_data,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=True)
])
@mock.patch.object(utils, 'ShallowMaskDict', autospec=True)
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_with_never_mask(self, mock_list, mock_apply_actions,
mock_check_conditions,
mock_get_built_in, mock_masked_dict):
"""Test apply_rules with mask_secrets='never'."""
self.config(mask_secrets='never', group='inspection_rules')
mock_list.return_value = [self.rule1]
mock_get_built_in.return_value = [self.rule2]
mock_check_conditions.return_value = True
masked_inventory = mock.MagicMock()
masked_plugin_data = mock.MagicMock()
mock_masked_dict.side_effect = [masked_inventory, masked_plugin_data,
mock.MagicMock(), mock.MagicMock()]
with task_manager.acquire(self.context, self.node.uuid) as task:
engine.apply_rules(task, self.inventory, self.plugin_data, 'main')
mock_masked_dict.assert_has_calls([
mock.call(self.inventory,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=False),
mock.call(self.plugin_data,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=False),
mock.call(self.inventory,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=False),
mock.call(self.plugin_data,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=False)
])
@mock.patch.object(utils, 'ShallowMaskDict', autospec=True)
@mock.patch.object(engine, 'get_built_in_rules', autospec=True)
@mock.patch.object(engine, 'check_conditions', autospec=True)
@mock.patch.object(engine, 'apply_actions', autospec=True)
@mock.patch('ironic.objects.InspectionRule.list', autospec=True)
def test_apply_rules_with_sensitive_mask(self, mock_list,
mock_apply_actions,
mock_check_conditions,
mock_get_built_in,
mock_masked_dict):
"""Test apply_rules with mask_secrets='sensitive'."""
self.config(mask_secrets='sensitive', group='inspection_rules')
mock_list.return_value = [self.rule1, self.sensitive_rule]
mock_get_built_in.return_value = []
mock_check_conditions.return_value = True
masked_inventory1 = mock.MagicMock()
masked_plugin_data1 = mock.MagicMock()
masked_inventory2 = mock.MagicMock()
masked_plugin_data2 = mock.MagicMock()
mock_masked_dict.side_effect = [
masked_inventory1, masked_plugin_data1,
masked_inventory2, masked_plugin_data2
]
with task_manager.acquire(self.context, self.node.uuid) as task:
engine.apply_rules(task, self.inventory, self.plugin_data, 'main')
mock_masked_dict.assert_has_calls([
mock.call(self.inventory,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=True),
mock.call(self.plugin_data,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=True),
mock.call(self.inventory,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=False),
mock.call(self.plugin_data,
sensitive_fields=engine.SENSITIVE_FIELDS,
mask_enabled=False)
])
class TestOperators(db_base.DbTestCase):
def setUp(self):
super(TestOperators, self).setUp()
self.node = obj_utils.create_test_node(self.context,
driver='fake-hardware')
def test_operator_exceptions(self):
"""Test that operators raise proper exceptions for invalid inputs."""
with task_manager.acquire(self.context, self.node.uuid) as task:
# NetOperator with invalid subnet
net_op = inspection_rules.operators.NetOperator()
self.assertRaises(
exception.InspectionRuleExecutionFailure,
net_op, task, address='192.168.1.1', subnet='invalid-subnet'
)
# MatchesOperator with invalid regex
matches_op = inspection_rules.operators.MatchesOperator()
self.assertRaises(
exception.InspectionRuleExecutionFailure,
matches_op, task, value='test', regex='[unclosed'
)
# ContainsOperator with invalid regex
contains_op = inspection_rules.operators.ContainsOperator()
self.assertRaises(
exception.InspectionRuleExecutionFailure,
contains_op, task, value='test', regex='[unclosed'
)
# SimpleOperator with non-list values
eq_op = inspection_rules.operators.EqOperator()
self.assertRaises(
exception.RuleConditionCheckFailure,
eq_op, task, values="not-a-list"
)
def test_oneofoperator_edge_cases(self):
"""Test OneOfOperator with edge cases."""
with task_manager.acquire(self.context, self.node.uuid) as task:
op = inspection_rules.operators.OneOfOperator()
self.assertFalse(op(task, value='test', values=[]))
self.assertFalse(op(task, value=None, values=['a', 'b']))
self.assertTrue(op(task, value='a', values=['a', 'b']))
def test_is_true_false_operators_edge_cases(self):
"""Test IsTrueOperator and IsFalseOperator."""
with task_manager.acquire(self.context, self.node.uuid) as task:
true_op = inspection_rules.operators.IsTrueOperator()
false_op = inspection_rules.operators.IsFalseOperator()
self.assertTrue(true_op(task, value='yes'))
self.assertTrue(true_op(task, value='TRUE'))
self.assertFalse(true_op(task, value='no'))
self.assertTrue(false_op(task, value='no'))
self.assertTrue(false_op(task, value='FALSE'))
self.assertFalse(false_op(task, value='yes'))
self.assertTrue(true_op(task, value=1))
self.assertTrue(true_op(task, value=0.1))
self.assertFalse(true_op(task, value=0))
self.assertTrue(false_op(task, value=0))
self.assertFalse(false_op(task, value=1))
self.assertFalse(true_op(task, value=None))
self.assertTrue(false_op(task, value=None))
self.assertFalse(true_op(task, value={}))
self.assertFalse(true_op(task, value=[]))
def test_operator_with_loop(self):
"""Test operator check_with_loop method."""
condition = {
'op': 'eq',
'loop': [
{'values': [1, 1]},
{'values': [2, 2]},
{'values': [3, 4]}
],
'multiple': 'any'
}
inventory = {'data': 'test'}
plugin_data = {'plugin': 'data'}
with task_manager.acquire(self.context, self.node.uuid) as task:
op = inspection_rules.operators.EqOperator()
# 'any' multiple (should return True)
self.assertTrue(op.check_with_loop(task, condition, inventory,
plugin_data))
# 'all' multiple (should return False)
condition['multiple'] = 'all'
self.assertFalse(op.check_with_loop(task, condition, inventory,
plugin_data))
# 'first' multiple (should return True)
condition['multiple'] = 'first'
self.assertTrue(op.check_with_loop(task, condition, inventory,
plugin_data))
# 'last' multiple (should return False)
condition['multiple'] = 'last'
self.assertFalse(op.check_with_loop(task, condition, inventory,
plugin_data))
def test_rule_operators(self):
"""Test all inspection_rules.operators with True and False cases."""
operator_tests = {
inspection_rules.operators.EqOperator: [
{'values': [5, 5]},
{'values': [5, 10]}
],
inspection_rules.operators.LtOperator: [
{'values': [5, 10]},
{'values': [10, 5]}
],
inspection_rules.operators.GtOperator: [
{'values': [10, 5]},
{'values': [5, 10]}
],
inspection_rules.operators.EmptyOperator: [
{'value': ''},
{'value': 'not empty'}
],
inspection_rules.operators.NetOperator: [
{'address': '192.168.1.5', 'subnet': '192.168.1.0/24'},
{'address': '10.0.0.1', 'subnet': '192.168.1.0/24'}
],
inspection_rules.operators.MatchesOperator: [
{'value': 'abc123', 'regex': r'abc\d+'},
{'value': 'xyz123', 'regex': r'abc\d+'}
],
inspection_rules.operators.ContainsOperator: [
{'value': 'test-abc123-end', 'regex': r'abc\d+'},
{'value': 'test-xyz-end', 'regex': r'abc\d+'}
],
inspection_rules.operators.OneOfOperator: [
{'value': 'b', 'values': ['a', 'b', 'c']},
{'value': 'z', 'values': ['a', 'b', 'c']}
],
inspection_rules.operators.IsNoneOperator: [
{'value': 'None'},
{'value': 'something'}
],
inspection_rules.operators.IsTrueOperator: [
{'value': True},
{'value': False}
],
inspection_rules.operators.IsFalseOperator: [
{'value': False},
{'value': True}
]
}
with task_manager.acquire(self.context, self.node.uuid) as task:
for op_class, test_cases in operator_tests.items():
op = op_class()
result = op(task, **test_cases[0])
self.assertTrue(result)
result = op(task, **test_cases[1])
self.assertFalse(result)
class TestActions(db_base.DbTestCase):
"""Test inspection rule actions"""
def setUp(self):
super(TestActions, self).setUp()
self.node = obj_utils.create_test_node(self.context,
driver='fake-hardware')
@mock.patch.object(inspection_rules.actions.LOG, 'info', autospec=True)
def test_log_action(self, mock_log):
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.LogAction()
test_msg = "Test log message"
action(task, msg=test_msg)
mock_log.assert_called_once_with(test_msg)
def test_fail_action(self):
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.FailAction()
error_msg = "Test failure"
self.assertRaises(exception.HardwareInspectionFailure,
action, task, msg=error_msg)
def test_set_attribute_action(self):
"""Test SetAttributeAction sets node attribute."""
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.SetAttributeAction()
action(task, path='extra', value={'test_key': 'test_value'})
self.assertEqual({'test_key': 'test_value'}, task.node.extra)
def test_extend_attribute_action(self):
"""Test ExtendAttributeAction extends a list attribute."""
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.ExtendAttributeAction()
task.node.tags = ['existing']
action(task, path='tags', value='new_tag')
self.assertEqual(['existing', 'new_tag'], task.node.tags)
def test_del_attribute_action(self):
"""Test DelAttributeAction deletes a node attribute."""
with task_manager.acquire(self.context, self.node.uuid) as task:
# Set up a value to delete
task.node.extra = {'to_delete': 'value'}
action = inspection_rules.actions.DelAttributeAction()
action(task, path='extra/to_delete')
self.assertEqual({}, task.node.extra)
@mock.patch.object(inspection_rules.actions.objects.Trait, 'create',
autospec=True)
def test_add_trait_action(self, mock_create):
"""Test AddTraitAction adds a node trait."""
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.AddTraitAction()
trait_name = 'CUSTOM_AWESOME_TRAIT'
action(task, name=trait_name)
mock_create.assert_called_once()
trait = mock_create.call_args[0][0]
self.assertEqual(trait_name, trait.trait)
@mock.patch.object(inspection_rules.actions.objects.Trait, 'destroy',
autospec=True)
def test_remove_trait_action(self, mock_destroy):
"""Test RemoveTraitAction removes a node trait."""
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.RemoveTraitAction()
trait_name = 'CUSTOM_AWESOME_TRAIT'
action(task, name=trait_name)
mock_destroy.assert_called_once_with(
task.context, node_id=task.node.id, trait=trait_name)
@mock.patch.object(inspection_rules.actions.driver_utils,
'add_node_capability', autospec=True)
def test_set_capability_action(self, mock_add):
"""Test SetCapabilityAction sets a node capability."""
with task_manager.acquire(self.context, self.node.uuid) as task:
action = inspection_rules.actions.SetCapabilityAction()
action(task, name='boot_mode', value='uefi')
mock_add.assert_called_once_with(task, 'boot_mode', 'uefi')
def test_unset_capability_action(self):
"""Test UnsetCapabilityAction removes a node capability."""
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.properties = {
'capabilities': 'boot_mode:uefi,other:value'}
action = inspection_rules.actions.UnsetCapabilityAction()
action(task, name='boot_mode')
self.assertEqual('other:value',
task.node.properties['capabilities'])
def test_set_port_attribute_action(self):
"""Test SetPortAttributeAction sets a port attribute."""
fake_port = mock.Mock()
fake_port.uuid = uuidutils.generate_uuid()
with task_manager.acquire(self.context, self.node.uuid) as task:
task.ports = [fake_port]
action = inspection_rules.actions.SetPortAttributeAction()
action(task, port_id=fake_port.uuid, path='extra',
value='test_value')
setattr(fake_port, 'extra', 'test_value')
fake_port.save.assert_called_once()
def test_extend_port_attribute_action(self):
"""Test ExtendPortAttributeAction extends a port attribute list."""
fake_port = mock.Mock()
fake_port.uuid = uuidutils.generate_uuid()
fake_port.tags = ['existing']
with task_manager.acquire(self.context, self.node.uuid) as task:
task.ports = [fake_port]
action = inspection_rules.actions.ExtendPortAttributeAction()
action(task, port_id=fake_port.uuid, path='tags', value='new_tag')
setattr(fake_port, 'tags', ['existing', 'new_tag'])
fake_port.save.assert_called_once()
def test_del_port_attribute_action(self):
"""Test DelPortAttributeAction deletes a port attribute."""
fake_port = mock.Mock()
fake_port.uuid = uuidutils.generate_uuid()
fake_port.extra = {'to_delete': 'value'}
with task_manager.acquire(self.context, self.node.uuid) as task:
task.ports = [fake_port]
action = inspection_rules.actions.DelPortAttributeAction()
action(task, port_id=fake_port.uuid, path='extra/to_delete')
fake_port.save.assert_called_once()
def test_set_plugin_data_action(self):
"""Test SetPluginDataAction sets plugin data."""
with task_manager.acquire(self.context, self.node.uuid) as task:
plugin_data = {'existing': 'data'}
action = inspection_rules.actions.SetPluginDataAction()
action(task, path='test_key', value='test_value',
plugin_data=plugin_data)
expected = {'existing': 'data', 'test_key': 'test_value'}
self.assertEqual(expected, plugin_data)
def test_extend_plugin_data_action(self):
"""Test ExtendPluginDataAction extends a plugin data list."""
with task_manager.acquire(self.context, self.node.uuid) as task:
plugin_data = {'test_list': ['item1']}
action = inspection_rules.actions.ExtendPluginDataAction()
action(task, path='test_list', value='item2',
plugin_data=plugin_data)
expected = {'test_list': ['item1', 'item2']}
self.assertEqual(expected, plugin_data)
def test_unset_plugin_data_action(self):
"""Test UnsetPluginDataAction removes plugin data."""
with task_manager.acquire(self.context, self.node.uuid) as task:
plugin_data = {'to_remove': 'value', 'keep': 'value'}
action = inspection_rules.actions.UnsetPluginDataAction()
action(task, path='to_remove', plugin_data=plugin_data)
self.assertEqual({'keep': 'value'}, plugin_data)
def test_action_error_cases(self):
"""Test that actions properly handle error cases."""
with task_manager.acquire(self.context, self.node.uuid) as task:
# SetAttributeAction nested path on a non-dict
set_attr = inspection_rules.actions.SetAttributeAction()
task.node.driver = 'fake-hardware'
self.assertRaises(
exception.RuleActionExecutionFailure,
set_attr, task, path='driver.some_key', value='test'
)
# ExtendAttributeAction non-list attribute
task.node.driver = 'fake-hardware'
extend_attr = inspection_rules.actions.ExtendAttributeAction()
self.assertRaises(
exception.RuleActionExecutionFailure,
extend_attr, task, path='driver', value='new_item'
)
# DelAttributeAction nested path on a non-dict
del_attr = inspection_rules.actions.DelAttributeAction()
task.node.driver = 'fake-hardware'
self.assertRaises(
exception.RuleActionExecutionFailure,
del_attr, task, path='driver.nonexistent_key'
)
# SetPortAttributeAction non-existent port
set_port = inspection_rules.actions.SetPortAttributeAction()
fake_port_id = uuidutils.generate_uuid()
self.assertRaises(
exception.PortNotFound,
set_port, task, port_id=fake_port_id, path='extra',
value='test'
)
# LogAction with invalid log level
log_action = inspection_rules.actions.LogAction()
self.assertRaises(
exception.InspectionRuleExecutionFailure,
log_action, task, msg='test message', level='invalid_level'
)
def test_action_with_loop(self):
"""Test action execute_with_loop method."""
action_data = {
'op': 'set-attribute',
'loop': [
{'path': 'extra/test1', 'value': 'value1'},
{'path': 'extra/test2', 'value': 'value2'}
]
}
inventory = {'data': 'test'}
plugin_data = {'plugin': 'data'}
with task_manager.acquire(self.context, self.node.uuid) as task:
task.node.extra = {}
# execute_with_loop
action = inspection_rules.actions.SetAttributeAction()
results = action.execute_with_loop(task, action_data, inventory,
plugin_data)
# verify both loop items were processed
self.assertEqual(2, len(results))
self.assertEqual('value1', task.node.extra['test1'])
self.assertEqual('value2', task.node.extra['test2'])
ironic/tests/unit/db/utils.py
+58
View File
@@ -27,6 +27,7 @@ from ironic.objects import chassis
from ironic.objects import conductor
from ironic.objects import deploy_template
from ironic.objects import firmware
from ironic.objects import inspection_rule as rule
from ironic.objects import node
from ironic.objects import node_history
from ironic.objects import node_inventory
@@ -822,3 +823,60 @@ def get_test_firmware_component_list():
{'component': 'BIOS', 'initial_version': 'v1.5.0',
'current_version': 'v1.5.0', 'last_version_flashed': None},
]
def get_test_inspection_rule(**kw):
default_uuid = uuidutils.generate_uuid()
return {
'version': kw.get('version', rule.InspectionRule.VERSION),
'uuid': kw.get('uuid', default_uuid),
'description': kw.get('description', 'an inspection rule'),
'sensitive': kw.get('sensitive', False),
'phase': kw.get('phase', 'main'),
'priority': kw.get('priority', 0),
'actions': kw.get('actions', [get_test_inspection_rule_action(
inspection_rule_id=kw.get('uuid'))]),
'conditions': kw.get('conditions', [
get_test_inspection_rule_condition(
inspection_rule_id=kw.get('uuid'))]),
}
def get_test_inspection_rule_action(**kw):
action = {
'op': kw.get('op', 'set-attribute'),
'args': kw.get('args', ["/driver", "idrac"]),
}
if 'id' in kw:
action['id'] = kw['id']
action.update({k: v for k, v in kw.items() if k not in action})
return action
def get_test_inspection_rule_condition(**kw):
condition = {
'op': kw.get('op', 'is-true'),
'args': kw.get('args', ["{node.auto_discovered}"]),
'multiple': kw.get('multiple', 'any'),
}
if 'id' in kw:
condition['id'] = kw['id']
condition.update({k: v for k, v in kw.items() if k not in condition})
return condition
def create_test_inspection_rule(**kw):
"""Create a inspection rule in the DB and return InspectionRule model.
:param kw: kwargs with overriding values for the inspection rule.
:returns: Test InspectionRule DB object.
"""
inspection_rule = get_test_inspection_rule(**kw)
dbapi = db_api.get_instance()
if 'uuid' not in kw:
del inspection_rule['uuid']
return dbapi.create_inspection_rule(inspection_rule)
ironic/tests/unit/objects/utils.py
+24
View File
@@ -459,3 +459,27 @@ def create_test_firmware_component(ctxt, **kw):
fw_cmp = get_test_firmware_component(ctxt, **kw)
fw_cmp.create()
return fw_cmp
def get_test_inspection_rule(ctxt, **kw):
"""Return an InpsectionRule object with appropriate attributes.
NOTE: The object leaves the attributes marked as changed, such
that a create() could be used to commit it to the DB.
"""
db_inspection_rule = db_utils.get_test_inspection_rule(**kw)
inspection_rule = objects.InspectionRule(ctxt)
for key in db_inspection_rule:
setattr(inspection_rule, key, db_inspection_rule[key])
return inspection_rule
def create_test_inspection_rule(ctxt, **kw):
"""Create and return a test inspection rule object.
NOTE: The object leaves the attributes marked as changed, such
that a create() could be used to commit it to the DB.
"""
inspection_rule = get_test_inspection_rule(ctxt, **kw)
inspection_rule.create()
return inspection_rule