本文整理汇总了Python中st2reactor.rules.enforcer.RuleEnforcer类的典型用法代码示例。如果您正苦于以下问题:Python RuleEnforcer类的具体用法?Python RuleEnforcer怎么用?Python RuleEnforcer使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了RuleEnforcer类的14个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_action_default_jinja_parameter_value_render_fail
def test_action_default_jinja_parameter_value_render_fail(self):
# Action parameter render failure should result in a failed execution
rule = self.models['rules']['rule_action_default_value_render_fail.yaml']
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, rule)
execution_db = enforcer.enforce()
self.assertTrue(execution_db is None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref, rule.ref)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].status,
RULE_ENFORCEMENT_STATUS_FAILED)
self.assertFalse(action_service.request.called)
self.assertTrue(action_service.create_request.called)
self.assertEqual(action_service.create_request.call_args[0][0].action,
'wolfpack.a2_default_value')
self.assertTrue(action_service.update_status.called)
self.assertEqual(action_service.update_status.call_args[1]['new_status'],
action_constants.LIVEACTION_STATUS_FAILED)
expected_msg = ('Failed to render parameter "arrtype": \'dict object\' has no '
'attribute \'arrtype_value\'')
result = action_service.update_status.call_args[1]['result']
self.assertEqual(result['error'], expected_msg)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].failure_reason,
expected_msg)
开发者ID:StackStorm,项目名称:st2,代码行数:30,代码来源:test_enforce.py
示例2: evaluate
def evaluate(self):
"""
Evaluate trigger instance against the rule.
:return: ``True`` if the rule matches, ``False`` otherwise.
:rtype: ``boolean``
"""
rule_db = self._get_rule_db()
trigger_instance_db, trigger_db = self._get_trigger_instance_db()
# The trigger check needs to be performed here as that is not performed
# by RulesMatcher.
if rule_db.trigger != trigger_db.ref:
LOG.info('rule.trigger "%s" and trigger.ref "%s" do not match.',
rule_db.trigger, trigger_db.ref)
return False
# Check if rule matches criteria.
matcher = RulesMatcher(trigger_instance=trigger_instance_db, trigger=trigger_db,
rules=[rule_db], extra_info=True)
matching_rules = matcher.get_matching_rules()
# Rule does not match so early exit.
if len(matching_rules) < 1:
return False
# Check if rule can be enforced
enforcer = RuleEnforcer(trigger_instance=trigger_instance_db, rule=rule_db)
runner_type_db = mock.Mock()
runner_type_db.runner_parameters = {}
action_db = mock.Mock()
action_db.parameters = {}
params = rule_db.action.parameters # pylint: disable=no-member
context, additional_contexts = enforcer.get_action_execution_context(action_db=action_db,
trace_context=None)
# Note: We only return partially resolved parameters.
# To be able to return all parameters we would need access to corresponding ActionDB,
# RunnerTypeDB and ConfigDB object, but this would add a dependency on the database and the
# tool is meant to be used standalone.
try:
params = enforcer.get_resolved_parameters(action_db=action_db,
runnertype_db=runner_type_db,
params=params,
context=context,
additional_contexts=additional_contexts)
LOG.info('Action parameters resolved to:')
for param in six.iteritems(params):
LOG.info('\t%s: %s', param[0], param[1])
return True
except (UndefinedError, ValueError) as e:
LOG.error('Failed to resolve parameters\n\tOriginal error : %s', six.text_type(e))
return False
except:
LOG.exception('Failed to resolve parameters.')
return False
开发者ID:nzlosh,项目名称:st2,代码行数:60,代码来源:tester.py
示例3: test_ruleenforcement_casts
def test_ruleenforcement_casts(self):
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, self.models['rules']['rule2.yaml'])
execution_db = enforcer.enforce()
self.assertTrue(execution_db is not None)
self.assertTrue(action_service.request.called)
self.assertTrue(isinstance(action_service.request.call_args[0][0].parameters['objtype'],
dict))
开发者ID:E-LLP,项目名称:st2,代码行数:7,代码来源:test_enforce.py
示例4: test_ruleenforcement_create_on_fail
def test_ruleenforcement_create_on_fail(self):
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, self.models['rules']['rule1.yaml'])
execution_db = enforcer.enforce()
self.assertTrue(execution_db is None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].failure_reason,
FAILURE_REASON)
开发者ID:E-LLP,项目名称:st2,代码行数:7,代码来源:test_enforce.py
示例5: test_ruleenforcement_create_on_success
def test_ruleenforcement_create_on_success(self):
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, self.models['rules']['rule2.yaml'])
execution_db = enforcer.enforce()
self.assertTrue(execution_db is not None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref,
self.models['rules']['rule2.yaml'].ref)
开发者ID:E-LLP,项目名称:st2,代码行数:7,代码来源:test_enforce.py
示例6: test_triggered_execution
def test_triggered_execution(self):
docs = {
'trigger_type': copy.deepcopy(fixture.ARTIFACTS['trigger_type']),
'trigger': copy.deepcopy(fixture.ARTIFACTS['trigger']),
'rule': copy.deepcopy(fixture.ARTIFACTS['rule']),
'trigger_instance': copy.deepcopy(fixture.ARTIFACTS['trigger_instance'])}
# Trigger an action execution.
trigger_type = TriggerType.add_or_update(
TriggerTypeAPI.to_model(TriggerTypeAPI(**docs['trigger_type'])))
trigger = Trigger.add_or_update(TriggerAPI.to_model(TriggerAPI(**docs['trigger'])))
rule = RuleAPI.to_model(RuleAPI(**docs['rule']))
rule.trigger = reference.get_str_resource_ref_from_model(trigger)
rule = Rule.add_or_update(rule)
trigger_instance = TriggerInstance.add_or_update(
TriggerInstanceAPI.to_model(TriggerInstanceAPI(**docs['trigger_instance'])))
trace_service.add_or_update_given_trace_context(
trace_context={'trace_tag': 'test_triggered_execution_trace'},
trigger_instances=[str(trigger_instance.id)])
enforcer = RuleEnforcer(trigger_instance, rule)
enforcer.enforce()
# Wait for the action execution to complete and then confirm outcome.
liveaction = LiveAction.get(context__trigger_instance__id=str(trigger_instance.id))
self.assertIsNotNone(liveaction)
liveaction = self._wait_on_status(liveaction, action_constants.LIVEACTION_STATUS_FAILED)
execution = self._get_action_execution(
liveaction__id=str(liveaction.id),
raise_exception=True
)
self.assertDictEqual(execution.trigger, vars(TriggerAPI.from_model(trigger)))
self.assertDictEqual(execution.trigger_type, vars(TriggerTypeAPI.from_model(trigger_type)))
self.assertDictEqual(execution.trigger_instance,
vars(TriggerInstanceAPI.from_model(trigger_instance)))
self.assertDictEqual(execution.rule, vars(RuleAPI.from_model(rule)))
action = action_utils.get_action_by_ref(liveaction.action)
self.assertDictEqual(execution.action, vars(ActionAPI.from_model(action)))
runner = RunnerType.get_by_name(action.runner_type['name'])
self.assertDictEqual(execution.runner, vars(RunnerTypeAPI.from_model(runner)))
liveaction = LiveAction.get_by_id(str(liveaction.id))
self.assertEqual(execution.start_timestamp, liveaction.start_timestamp)
self.assertEqual(execution.end_timestamp, liveaction.end_timestamp)
self.assertEqual(execution.result, liveaction.result)
self.assertEqual(execution.status, liveaction.status)
self.assertEqual(execution.context, liveaction.context)
self.assertEqual(execution.liveaction['callback'], liveaction.callback)
self.assertEqual(execution.liveaction['action'], liveaction.action)
开发者ID:StackStorm,项目名称:st2,代码行数:49,代码来源:test_executions.py
示例7: assertResolvedParamsMatchExpected
def assertResolvedParamsMatchExpected(self, rule, trigger_instance, params, expected_params):
runner_type_db = mock.Mock()
runner_type_db.runner_parameters = {}
action_db = mock.Mock()
action_db.parameters = {}
enforcer = RuleEnforcer(trigger_instance, rule)
context, additional_contexts = enforcer.get_action_execution_context(action_db=action_db)
resolved_params = enforcer.get_resolved_parameters(action_db=action_db,
runnertype_db=runner_type_db,
params=params,
context=context,
additional_contexts=additional_contexts)
self.assertEqual(resolved_params, expected_params)
开发者ID:StackStorm,项目名称:st2,代码行数:15,代码来源:test_enforce.py
示例8: test_action_default_jinja_parameter_value_overridden_in_rule
def test_action_default_jinja_parameter_value_overridden_in_rule(self):
# Verify that it works correctly if default parameter value is overridden in rule
rule = self.models['rules']['rule_action_default_value_overridden.yaml']
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, rule)
execution_db = enforcer.enforce()
self.assertTrue(execution_db is not None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref, rule.ref)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].status,
RULE_ENFORCEMENT_STATUS_SUCCEEDED)
call_parameters = action_service.request.call_args[0][0].parameters
self.assertEqual(call_parameters['objtype'], {'t1_p': 't1_p_v'})
self.assertEqual(call_parameters['strtype'], 't1_p_v')
self.assertEqual(call_parameters['arrtype'], ['override 1', 'override 2'])
开发者ID:StackStorm,项目名称:st2,代码行数:18,代码来源:test_enforce.py
示例9: test_action_default_jinja_parameter_value_is_rendered
def test_action_default_jinja_parameter_value_is_rendered(self):
# Verify that a default action parameter which is a Jinja variable is correctly rendered
rule = self.models['rules']['rule_action_default_value.yaml']
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, rule)
execution_db = enforcer.enforce()
self.assertTrue(execution_db is not None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref, rule.ref)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].status,
RULE_ENFORCEMENT_STATUS_SUCCEEDED)
call_parameters = action_service.request.call_args[0][0].parameters
self.assertEqual(call_parameters['objtype'], {'t1_p': 't1_p_v'})
self.assertEqual(call_parameters['strtype'], 't1_p_v')
self.assertEqual(call_parameters['arrtype'], ['one 1', 'two 2', 'three 3'])
开发者ID:StackStorm,项目名称:st2,代码行数:18,代码来源:test_enforce.py
示例10: evaluate
def evaluate(self):
"""
Evaluate trigger instance against the rule.
:return: ``True`` if the rule matches, ``False`` otherwise.
:rtype: ``boolean``
"""
rule_db = self._get_rule_db()
trigger_instance_db, trigger_db = self._get_trigger_instance_db()
# The trigger check needs to be performed here as that is not performed
# by RulesMatcher.
if rule_db.trigger != trigger_db.ref:
LOG.info('rule.trigger "%s" and trigger.ref "%s" do not match.', rule_db.trigger, trigger_db.ref)
return False
# Check if rule matches criteria.
matcher = RulesMatcher(
trigger_instance=trigger_instance_db, trigger=trigger_db, rules=[rule_db], extra_info=True
)
matching_rules = matcher.get_matching_rules()
# Rule does not match so early exit.
if len(matching_rules) < 1:
return False
# Check if rule can be enforced
try:
enforcer = RuleEnforcer(trigger_instance=trigger_instance_db, rule=rule_db)
params = enforcer.get_resolved_parameters()
LOG.info("Action parameters resolved to:")
for param in six.iteritems(params):
LOG.info("\t%s: %s", param[0], param[1])
return True
except (UndefinedError, ValueError) as e:
LOG.error("Failed to resolve parameters\n\tOriginal error : %s", str(e))
return False
except:
LOG.exception("Failed to resolve parameters.")
return False
开发者ID:rlugojr,项目名称:st2,代码行数:41,代码来源:tester.py
示例11: test_triggered_execution
def test_triggered_execution(self):
docs = {
'trigger_type': copy.deepcopy(fixture.ARTIFACTS['trigger_type']),
'trigger': copy.deepcopy(fixture.ARTIFACTS['trigger']),
'rule': copy.deepcopy(fixture.ARTIFACTS['rule']),
'trigger_instance': copy.deepcopy(fixture.ARTIFACTS['trigger_instance'])}
# Trigger an action execution.
trigger_type = TriggerType.add_or_update(
TriggerTypeAPI.to_model(TriggerTypeAPI(**docs['trigger_type'])))
trigger = Trigger.add_or_update(TriggerAPI.to_model(TriggerAPI(**docs['trigger'])))
rule = RuleAPI.to_model(RuleAPI(**docs['rule']))
rule.trigger = reference.get_str_resource_ref_from_model(trigger)
rule = Rule.add_or_update(rule)
trigger_instance = TriggerInstance.add_or_update(
TriggerInstanceAPI.to_model(TriggerInstanceAPI(**docs['trigger_instance'])))
enforcer = RuleEnforcer(trigger_instance, rule)
enforcer.enforce()
# Wait for the action execution to complete and then confirm outcome.
execution = ActionExecution.get(context__trigger_instance__id=str(trigger_instance.id))
self.assertIsNotNone(execution)
execution = ActionExecution.get_by_id(str(execution.id))
self.assertEqual(execution.status, ACTIONEXEC_STATUS_SUCCEEDED)
history = ActionExecutionHistory.get(execution__id=str(execution.id), raise_exception=True)
self.assertDictEqual(history.trigger, vars(TriggerAPI.from_model(trigger)))
self.assertDictEqual(history.trigger_type, vars(TriggerTypeAPI.from_model(trigger_type)))
self.assertDictEqual(history.trigger_instance,
vars(TriggerInstanceAPI.from_model(trigger_instance)))
self.assertDictEqual(history.rule, vars(RuleAPI.from_model(rule)))
action_ref = ResourceReference.from_string_reference(ref=execution.action)
action, _ = action_utils.get_action_by_dict(
{'name': action_ref.name, 'pack': action_ref.pack})
self.assertDictEqual(history.action, vars(ActionAPI.from_model(action)))
runner = RunnerType.get_by_name(action.runner_type['name'])
self.assertDictEqual(history.runner, vars(RunnerTypeAPI.from_model(runner)))
execution = ActionExecution.get_by_id(str(execution.id))
self.assertDictEqual(history.execution, vars(ActionExecutionAPI.from_model(execution)))
开发者ID:bjoernbessert,项目名称:st2,代码行数:38,代码来源:test_history.py
示例12: test_ruleenforcement_occurs
def test_ruleenforcement_occurs(self):
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, MOCK_RULE_1)
execution_id = enforcer.enforce()
self.assertTrue(execution_id is not None)
开发者ID:bjoernbessert,项目名称:st2,代码行数:4,代码来源:test_enforce.py
示例13: test_ruleenforcement_occurs
def test_ruleenforcement_occurs(self):
enforcer = RuleEnforcer(MOCK_TRIGGER_INSTANCE, self.models['rules']['rule1.yaml'])
execution_db = enforcer.enforce()
self.assertTrue(execution_db is not None)
开发者ID:E-LLP,项目名称:st2,代码行数:4,代码来源:test_enforce.py
示例14: test_rule_enforcement_create_rule_none_param_casting
def test_rule_enforcement_create_rule_none_param_casting(self):
mock_trigger_instance = MOCK_TRIGGER_INSTANCE_2
# 1. Non None value, should be serialized as regular string
mock_trigger_instance.payload = {'t1_p': 'somevalue'}
def mock_cast_string(x):
assert x == 'somevalue'
return casts._cast_string(x)
casts.CASTS['string'] = mock_cast_string
enforcer = RuleEnforcer(mock_trigger_instance,
self.models['rules']['rule_use_none_filter.yaml'])
execution_db = enforcer.enforce()
# Verify value has been serialized correctly
call_args = action_service.request.call_args[0]
live_action_db = call_args[0]
self.assertEqual(live_action_db.parameters['actionstr'], 'somevalue')
self.assertTrue(execution_db is not None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref,
self.models['rules']['rule_use_none_filter.yaml'].ref)
# 2. Verify that None type from trigger instance is correctly serialized to
# None when using "use_none" Jinja filter when invoking an action
mock_trigger_instance.payload = {'t1_p': None}
def mock_cast_string(x):
assert x == NONE_MAGIC_VALUE
return casts._cast_string(x)
casts.CASTS['string'] = mock_cast_string
enforcer = RuleEnforcer(mock_trigger_instance,
self.models['rules']['rule_use_none_filter.yaml'])
execution_db = enforcer.enforce()
# Verify None has been correctly serialized to None
call_args = action_service.request.call_args[0]
live_action_db = call_args[0]
self.assertEqual(live_action_db.parameters['actionstr'], None)
self.assertTrue(execution_db is not None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref,
self.models['rules']['rule_use_none_filter.yaml'].ref)
casts.CASTS['string'] = casts._cast_string
# 3. Parameter value is a compound string one of which values is None, but "use_none"
# filter is not used
mock_trigger_instance = MOCK_TRIGGER_INSTANCE_3
mock_trigger_instance.payload = {'t1_p': None, 't2_p': 'value2'}
enforcer = RuleEnforcer(mock_trigger_instance,
self.models['rules']['rule_none_no_use_none_filter.yaml'])
execution_db = enforcer.enforce()
# Verify None has been correctly serialized to None
call_args = action_service.request.call_args[0]
live_action_db = call_args[0]
self.assertEqual(live_action_db.parameters['actionstr'], 'None-value2')
self.assertTrue(execution_db is not None)
self.assertTrue(RuleEnforcement.add_or_update.called)
self.assertEqual(RuleEnforcement.add_or_update.call_args[0][0].rule.ref,
self.models['rules']['rule_none_no_use_none_filter.yaml'].ref)
casts.CASTS['string'] = casts._cast_string
开发者ID:E-LLP,项目名称:st2,代码行数:67,代码来源:test_enforce.py
注:本文中的st2reactor.rules.enforcer.RuleEnforcer类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论