openstack
/
mistral
Code Issues Proposed changes

Allowing strings in on-success/on-error/on-complete clauses 

Change-Id: Ib34ee639d9aae9f7ff8cd685963efa7bf43d6556
(cherry picked from commit 7e25126039)
This commit is contained in:
Renat Akhmerov 2015-04-21 15:02:12 +06:00
parent e0a4654ab0
commit 7552bfeab9
9 changed files with 119 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
AUTHORS
View File

@ -5,12 +5,12 @@ Angus Salkeld <angus.salkeld@rackspace.com>
Ankita Wagh <ankita_wagh@symmactoolkit-c02lr80ufd57.symc.symantec.com>
Boris Pavlovic <boris@pavlovic.me>
Christian Berendt <berendt@b1-systems.de>
David C Kennedy <david.c.kennedy@hp.com>
Dmitri Zimine <dz@stackstorm.com>
Jeremy Stanley <fungi@yuggoth.org>
Kirill Izotov <enykeev@stackstorm.com>
Lakshmi Kannan <lakshmi@stackstorm.com>
Lingxian Kong <anlin.kong@gmail.com>
Limor Stotland <limor.bortman@alcatel-lucent.com>
Lingxian Kong <konglingxian@huawei.com>
Manas Kelshikar <manas@stackstorm.com>
Nikolay Mahotkin <nmakhotkin@mirantis.com>
Pierre-Arthur MATHIEU <pierre-arthur.mathieu@hp.com>
@ -22,5 +22,5 @@ Timur Nurlygayanov <tnurlygayanov@mirantis.com>
Winson Chan <wcchan@stackstorm.com>
ZhiQiang Fan <zhiqiang.fan@huawei.com>
Bryan Havenstein <bryan.havenstein@ericsson.com>
David Kennedy <dkennedy@hp.com>
Liu Sheng <liusheng@huawei.com>

2
mistral/engine/default_engine.py
View File

@ -100,7 +100,7 @@ class DefaultEngine(base.Engine):
# Must be before loading the object itself (see method doc).
self._lock_workflow_execution(wf_ex_id)
wf_ex = db_api.get_workflow_execution(wf_ex_id)
wf_ex = task_ex.workflow_execution
wf_trace.info(
task_ex,

12
mistral/tests/unit/engine/base.py
View File

@ -111,15 +111,21 @@ class EngineTestCase(base.DbTestCase):
def print_workflow_executions(self, exc_info):
print("\nEngine test case exception occurred: %s" % exc_info[1])
print("Exception type: %s" % exc_info[0])
print("\nPrinting failed workflows...")
print("\nPrinting workflow executions...")
wf_execs = db_api.get_workflow_executions()
for wf_ex in wf_execs:
print("\n%s [state=%s]" % (wf_ex.name, wf_ex.state))
print(
"\n%s [state=%s, output=%s]" %
(wf_ex.name, wf_ex.state, wf_ex.output)
)
for t_ex in wf_ex.task_executions:
print("\t%s [state=%s]" % (t_ex.name, t_ex.state))
print(
"\t%s [state=%s, published=%s]" %
(t_ex.name, t_ex.state, t_ex.published)
)
def is_task_in_state(self, task_ex_id, state):
return db_api.get_task_execution(task_ex_id).state == state

67
mistral/tests/unit/engine/test_direct_workflow.py
View File

@ -23,7 +23,6 @@ from mistral.services import workflows as wf_service
from mistral.tests.unit.engine import base
from mistral.workflow import states
# TODO(nmakhotkin) Need to write more tests.
LOG = logging.getLogger(__name__)
@ -44,7 +43,7 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
return db_api.get_workflow_execution(wf_ex.id)
def test_direct_workflow_on_closures(self):
wf = """
wf_text = """
version: '2.0'
wf:
@ -77,7 +76,7 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
action: std.noop
"""
wf_ex = self._run_workflow(wf)
wf_ex = self._run_workflow(wf_text)
tasks = wf_ex.task_executions
@ -94,7 +93,7 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
self.assertTrue(wf_ex.state, states.ERROR)
def test_wrong_task_input(self):
wf_wrong_task_input = """
wf_text = """
version: '2.0'
wf:
@ -111,7 +110,7 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
action: std.echo wrong_input="Hahaha"
"""
wf_ex = self._run_workflow(wf_wrong_task_input)
wf_ex = self._run_workflow(wf_text)
task_ex = self._assert_single_item(wf_ex.task_executions, name='task2')
action_ex = db_api.get_action_executions(
@ -131,7 +130,7 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
self.assertIn(action_ex.output['result'], wf_ex.state_info)
def test_wrong_first_task_input(self):
wf_invalid_first_task_input = """
wf_text = """
version: '2.0'
wf:
@ -142,7 +141,7 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
action: std.echo wrong_input="Ha-ha"
"""
wf_ex = self._run_workflow(wf_invalid_first_task_input)
wf_ex = self._run_workflow(wf_text)
task_ex = wf_ex.task_executions[0]
action_ex = db_api.get_action_executions(
@ -162,11 +161,12 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
self.assertIn(action_ex.output['result'], wf_ex.state_info)
def test_wrong_action(self):
wf_invalid_action = """
wf_text = """
version: '2.0'
wf:
type: direct
tasks:
task1:
action: std.echo output="Echo"
@ -176,7 +176,8 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
task2:
action: action.doesnt_exist
"""
wf_ex = self._run_workflow(wf_invalid_action)
wf_ex = self._run_workflow(wf_text)
# TODO(dzimine): Catch tasks caused error, and set them to ERROR:
# TODO(dzimine): self.assertTrue(task_ex.state, states.ERROR)
@ -185,16 +186,19 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
self.assertIn("Failed to find action", wf_ex.state_info)
def test_wrong_action_first_task(self):
wf_invalid_action_first_task = """
wf_text = """
version: '2.0'
wf:
type: direct
tasks:
task1:
action: wrong.task
"""
wf_service.create_workflows(wf_invalid_action_first_task)
wf_service.create_workflows(wf_text)
with mock.patch.object(de.DefaultEngine, '_fail_workflow') as mock_fw:
self.assertRaises(
exc.InvalidActionException,
@ -210,11 +214,12 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
)
def test_messed_yaql(self):
wf_messed_yaql = """
wf_text = """
version: '2.0'
wf:
type: direct
tasks:
task1:
action: std.echo output="Echo"
@ -224,12 +229,13 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
task2:
action: std.echo output=<% wrong(yaql) %>
"""
wf_ex = self._run_workflow(wf_messed_yaql)
wf_ex = self._run_workflow(wf_text)
self.assertTrue(wf_ex.state, states.ERROR)
def test_messed_yaql_in_first_task(self):
wf_messed_yaql_in_first_task = """
wf_text = """
version: '2.0'
wf:
@ -238,7 +244,8 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
task1:
action: std.echo output=<% wrong(yaql) %>
"""
wf_service.create_workflows(wf_messed_yaql_in_first_task)
wf_service.create_workflows(wf_text)
with mock.patch.object(de.DefaultEngine, '_fail_workflow') as mock_fw:
self.assertRaises(
@ -254,3 +261,33 @@ class DirectWorkflowEngineTest(base.EngineTestCase):
),
"Called with a right exception"
)
def test_one_line_syntax_in_on_clauses(self):
wf_text = """
version: '2.0'
wf:
type: direct
tasks:
task1:
action: std.echo output=1
on-success: task2
task2:
action: std.echo output=1
on-complete: task3
task3:
action: std.fail
on-error: task4
task4:
action: std.echo output=4
"""
wf_service.create_workflows(wf_text)
wf_ex = self.engine.start_workflow('wf', {})
self._await(lambda: self.is_execution_success(wf_ex.id))

8
mistral/tests/unit/workbook/v2/test_tasks.py
View File

@ -210,7 +210,7 @@ class TaskSpecValidation(v2_base.WorkflowSpecValidationTestCase):
({'on-success': [{'email': '<% 1 %>'}, 'echo']}, False),
({'on-success': [{'email': '<% $.v1 in $.v2 %>'}]}, False),
({'on-success': [{'email': '<% * %>'}]}, True),
({'on-success': 'email'}, True),
({'on-success': 'email'}, False),
({'on-success': None}, True),
({'on-success': ['']}, True),
({'on-success': []}, True),
@ -221,7 +221,7 @@ class TaskSpecValidation(v2_base.WorkflowSpecValidationTestCase):
({'on-error': [{'email': '<% 1 %>'}, 'echo']}, False),
({'on-error': [{'email': '<% $.v1 in $.v2 %>'}]}, False),
({'on-error': [{'email': '<% * %>'}]}, True),
({'on-error': 'email'}, True),
({'on-error': 'email'}, False),
({'on-error': None}, True),
({'on-error': ['']}, True),
({'on-error': []}, True),
@ -232,7 +232,7 @@ class TaskSpecValidation(v2_base.WorkflowSpecValidationTestCase):
({'on-complete': [{'email': '<% 1 %>'}, 'echo']}, False),
({'on-complete': [{'email': '<% $.v1 in $.v2 %>'}]}, False),
({'on-complete': [{'email': '<% * %>'}]}, True),
({'on-complete': 'email'}, True),
({'on-complete': 'email'}, False),
({'on-complete': None}, True),
({'on-complete': ['']}, True),
({'on-complete': []}, True),
@ -242,7 +242,9 @@ class TaskSpecValidation(v2_base.WorkflowSpecValidationTestCase):
for transition, expect_error in tests:
overlay = {'test': {'tasks': {}}}
utils.merge_dicts(overlay['test']['tasks'], {'get': transition})
self._parse_dsl_spec(add_tasks=True,
changes=overlay,
expect_error=expect_error)

8
mistral/tests/unit/workbook/v2/test_workflows.py
View File

@ -190,7 +190,7 @@ class WorkflowSpecValidation(base.WorkflowSpecValidationTestCase):
({'on-success': [{'email': '<% 1 %>'}, 'echo']}, False),
({'on-success': [{'email': '<% $.v1 in $.v2 %>'}]}, False),
({'on-success': [{'email': '<% * %>'}]}, True),
({'on-success': 'email'}, True),
({'on-success': 'email'}, False),
({'on-success': None}, True),
({'on-success': ['']}, True),
({'on-success': []}, True),
@ -201,7 +201,7 @@ class WorkflowSpecValidation(base.WorkflowSpecValidationTestCase):
({'on-error': [{'email': '<% 1 %>'}, 'echo']}, False),
({'on-error': [{'email': '<% $.v1 in $.v2 %>'}]}, False),
({'on-error': [{'email': '<% * %>'}]}, True),
({'on-error': 'email'}, True),
({'on-error': 'email'}, False),
({'on-error': None}, True),
({'on-error': ['']}, True),
({'on-error': []}, True),
@ -212,7 +212,7 @@ class WorkflowSpecValidation(base.WorkflowSpecValidationTestCase):
({'on-complete': [{'email': '<% 1 %>'}, 'echo']}, False),
({'on-complete': [{'email': '<% $.v1 in $.v2 %>'}]}, False),
({'on-complete': [{'email': '<% * %>'}]}, True),
({'on-complete': 'email'}, True),
({'on-complete': 'email'}, False),
({'on-complete': None}, True),
({'on-complete': ['']}, True),
({'on-complete': []}, True),
@ -260,7 +260,9 @@ class WorkflowSpecValidation(base.WorkflowSpecValidationTestCase):
for default, expect_error in tests:
overlay = {'test': {'task-defaults': {}}}
utils.merge_dicts(overlay['test']['task-defaults'], default)
self._parse_dsl_spec(add_tasks=True,
changes=overlay,
expect_error=expect_error)

3
mistral/workbook/base.py
View File

@ -151,6 +151,9 @@ class BaseSpec(object):
if not prop_val:
return []
if isinstance(prop_val, six.string_types):
return [self._as_tuple(prop_val)]
return [self._as_tuple(item) for item in prop_val]
@staticmethod

29
mistral/workbook/v2/task_defaults.py
View File

@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import six
from mistral.workbook import types
from mistral.workbook.v2 import base
from mistral.workbook.v2 import policies
@ -23,6 +25,13 @@ class TaskDefaultsSpec(base.BaseSpec):
_task_policies_schema = policies.PoliciesSpec.get_schema(
includes=None)
_on_clause_type = {
"oneOf": [
types.NONEMPTY_STRING,
types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST
]
}
_schema = {
"type": "object",
"properties": {
@ -32,9 +41,9 @@ class TaskDefaultsSpec(base.BaseSpec):
"timeout": policies.TIMEOUT_SCHEMA,
"pause-before": policies.PAUSE_BEFORE_SCHEMA,
"concurrency": policies.CONCURRENCY_SCHEMA,
"on-complete": types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST,
"on-success": types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST,
"on-error": types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST
"on-complete": _on_clause_type,
"on-success": _on_clause_type,
"on-error": _on_clause_type
},
"additionalProperties": False
}
@ -63,11 +72,15 @@ class TaskDefaultsSpec(base.BaseSpec):
super(TaskDefaultsSpec, self).validate()
# Validate YAQL expressions.
[self.validate_yaql_expr(transition)
for transition in (self._data.get('on-complete', []) +
self._data.get('on-success', []) +
self._data.get('on-error', []))
if isinstance(transition, dict)]
self._validate_transitions('on-complete')
self._validate_transitions('on-success')
self._validate_transitions('on-error')
def _validate_transitions(self, on_clause):
val = self._data.get(on_clause, [])
[self.validate_yaql_expr(t)
for t in ([val] if isinstance(val, six.string_types) else val)]
def get_policies(self):
return self._policies

27
mistral/workbook/v2/tasks.py
View File

@ -200,6 +200,13 @@ class TaskSpec(base.BaseSpec):
class DirectWorkflowTaskSpec(TaskSpec):
_type = 'direct'
_on_clause_type = {
"oneOf": [
types.NONEMPTY_STRING,
types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST
]
}
_direct_workflow_schema = {
"type": "object",
"properties": {
@ -210,9 +217,9 @@ class DirectWorkflowTaskSpec(TaskSpec):
types.POSITIVE_INTEGER
]
},
"on-complete": types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST,
"on-success": types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST,
"on-error": types.UNIQUE_STRING_OR_YAQL_CONDITION_LIST
"on-complete": _on_clause_type,
"on-success": _on_clause_type,
"on-error": _on_clause_type
}
}
@ -239,11 +246,15 @@ class DirectWorkflowTaskSpec(TaskSpec):
raise exc.InvalidModelException(msg)
# Validate YAQL expressions.
[self.validate_yaql_expr(transition)
for transition in (self._data.get('on-complete', []) +
self._data.get('on-success', []) +
self._data.get('on-error', []))
if isinstance(transition, dict)]
self._validate_transitions('on-complete')
self._validate_transitions('on-success')
self._validate_transitions('on-error')
def _validate_transitions(self, on_clause):
val = self._data.get(on_clause, [])
[self.validate_yaql_expr(t)
for t in ([val] if isinstance(val, six.string_types) else val)]
def get_join(self):
return self._join