Skip to content
Browse files

Add 'Execute' Task Spec

  • Loading branch information...
1 parent 6cb935b commit 1c96c12eac0f8fcdf9e25cebb042087209485e9e @ziadsawalha committed Apr 5, 2012
View
4 README
@@ -62,6 +62,10 @@ Hint: The examples are located in tests/data/spiff/.
10. Block Task to Sub-Workflow Decomposition [data/block_to_subworkflow.xml]
11. Sub-Workflow Decomposition to Block Task [data/subworkflow_to_block.xml]
+ Other Patterns:
+
+ 1. Execute - spawns a subprocess and waits for the results
+
Contact
--------
View
85 SpiffWorkflow/specs/Execute.py
@@ -0,0 +1,85 @@
+# Copyright (C) 2007 Samuel Abels
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+import subprocess
+
+from SpiffWorkflow.Task import Task
+from SpiffWorkflow.exceptions import WorkflowException
+from SpiffWorkflow.specs.TaskSpec import TaskSpec
+
+
+class Execute(TaskSpec):
+ """
+ This class executes an external process, goes into WAITING until the
+ process is complete, and returns the results of the execution.
+
+ Usage:
+
+ task = Execute(spec, 'Ping', args=["ping", "-t", "1", "127.0.0.1"])
+ ... when workflow complete
+ print workflow.get_task('Ping').results
+ """
+
+ def __init__(self, parent, name, args=None, **kwargs):
+ """
+ Constructor.
+
+ @type parent: TaskSpec
+ @param parent: A reference to the parent task spec.
+ @type name: str
+ @param name: The name of the task spec.
+ @type args: list
+ @param args: args to pass to process (first arg is the command).
+ @type kwargs: dict
+ @param kwargs: kwargs to pass-through to TaskSpec initializer.
+ """
+ assert parent is not None
+ assert name is not None
+ TaskSpec.__init__(self, parent, name, **kwargs)
+ self.args = args
+
+ def try_fire(self, my_task, force = False):
+ """Returns False when successfully fired, True otherwise"""
+ if (not hasattr(my_task, 'subprocess')) or my_task.subprocess is None:
+ my_task.subprocess = subprocess.Popen(self.args,
+ stderr=subprocess.STDOUT,
+ stdout=subprocess.PIPE)
+
+ if my_task.subprocess:
+ my_task.subprocess.poll()
+ if my_task.subprocess.returncode is None:
+ # Still waiting
+ return False
+ else:
+ results = my_task.subprocess.communicate()
+ my_task.results = results
+ return True
+ return False
+
+ def _update_state_hook(self, my_task):
+ if not self.try_fire(my_task):
+ my_task.state = Task.WAITING
+ result = False
+ else:
+ result = super(Execute, self)._update_state_hook(my_task)
+ return result
+
+ def serialize(self, serializer):
+ return serializer._serialize_execute(self)
+
+ @classmethod
+ def deserialize(self, serializer, wf_spec, s_state):
+ spec = serializer._deserialize_execute(wf_spec, s_state)
+ return spec
View
1 SpiffWorkflow/specs/__init__.py
@@ -3,6 +3,7 @@
from CancelTask import CancelTask
from Choose import Choose
from ExclusiveChoice import ExclusiveChoice
+from Execute import Execute
from Gate import Gate
from Join import Join
from MultiChoice import MultiChoice
View
10 SpiffWorkflow/storage/DictionarySerializer.py
@@ -183,6 +183,16 @@ def _deserialize_exclusive_choice(self, wf_spec, s_state):
spec.default_task_spec = s_state['default_task_spec']
return spec
+ def _serialize_execute(self, spec):
+ s_state = self._serialize_task_spec(spec)
+ s_state['args'] = spec.args
+ return s_state
+
+ def _deserialize_execute(self, wf_spec, s_state):
+ spec = Execute(wf_spec, s_state['name'], s_state['args'])
+ self._deserialize_task_spec(wf_spec, s_state, spec = spec)
+ return spec
+
def _serialize_gate(self, spec):
s_state = self._serialize_task_spec(spec)
s_state['context'] = spec.context
View
33 tests/SpiffWorkflow/TaskTest.py
@@ -1,9 +1,11 @@
+import time
import sys, unittest, re, os.path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..'))
from SpiffWorkflow import Task
-from SpiffWorkflow.Workflow import TaskIdAssigner
-from SpiffWorkflow.specs import WorkflowSpec, Simple
+from SpiffWorkflow.Workflow import TaskIdAssigner, Workflow
+from SpiffWorkflow.specs import WorkflowSpec, Simple, Execute
+from SpiffWorkflow.specs.TaskSpec import TaskSpec
from SpiffWorkflow.exceptions import WorkflowException
class MockWorkflow(object):
@@ -69,6 +71,33 @@ def testTree(self):
'Expected:\n' + expected2 + '\n' + \
'but got:\n' + result)
+ def test_execute(self):
+ """Tests that we can create a task that executes an shell command
+ and that the workflow can be called to complete such tasks"""
+ spec = WorkflowSpec()
+ task1 = Execute(spec, 'Ping', args=["ping", "-t", "1", "127.0.0.1"])
+ spec.start.connect(task1)
+ workflow = Workflow(spec)
+
+ i = 0
+ while not workflow.is_completed() and i < 10:
+ workflow.complete_all()
+ i += 1
+ time.sleep(0.5)
+ self.assertTrue(workflow.is_completed())
+ self.assertEqual(i, 3)
+ task = workflow.get_task(3)
+ self.assertEquals(task.state_history, [1, 8, 16, 64])
+ # Check whether the status log is accurate.
+ expected = """Moving 'Ping' from FUTURE to WAITING
+Moving 'Ping' from WAITING to READY
+Moving 'Ping' from READY to COMPLETED"""
+ self.assert_(expected == '\n'.join(task.log),
+ 'Expected:\n' + expected + '\n' + \
+ 'but got:\n' + '\n'.join(task.log))
+ self.assertIn('127.0.0.1', task.results[0])
+
+
def suite():
return unittest.TestLoader().loadTestsFromTestCase(TaskTest)
if __name__ == '__main__':

0 comments on commit 1c96c12

Please sign in to comment.
Something went wrong with that request. Please try again.