GitLab will be upgraded on June 23rd evening. During the upgrade the service will be unavailable, sorry for the inconvenience.

Commit 3cbe74f2 authored by Olof Svensson's avatar Olof Svensson

New tests with submodels

parent ca39d4bd
......@@ -39,9 +39,10 @@ class Workflow1(Workflow):
def __init__(self, name):
Workflow.__init__(self, name)
self.startActor = StartActor()
script = 'pypushflow.test.pythonActorTest.py'
name = 'Python Actor Test'
self.pythonActor = PythonActor(script=script, name=name)
self.pythonActor = PythonActor(
script='pypushflow.test.pythonActorTest.py',
name='Python Actor Test'
)
self.stopActor = StopActor()
self.startActor.connect(self.pythonActor)
self.pythonActor.connect(self.stopActor)
......@@ -63,9 +64,11 @@ class Workflow2(Workflow):
def __init__(self, name):
Workflow.__init__(self, name)
self.startActor = StartActor()
script = 'pypushflow.test.pythonErrorHandlerTest.py'
name = 'Python Error Handler Test'
self.pythonActor = PythonActor(script=script, name=name, errorHandler=self)
self.pythonActor = PythonActor(
script='pypushflow.test.pythonErrorHandlerTest.py',
name='Python Error Handler Test',
errorHandler=self
)
self.stopActor = StopActor()
self.startActor.connect(self.pythonActor)
self.pythonActor.connect(self.stopActor)
......@@ -83,18 +86,66 @@ class Submodel1(Submodel):
"""
def __init__(self, name):
Submodel.__init__(name=name)
script = 'pypushflow.test.pythonActorTest.py'
name = 'Python Actor Test'
self.pythonActor = PythonActor(script=script, name=name)
Submodel.__init__(self, name=name)
self.pythonActor = PythonActor(
script='pypushflow.test.pythonActorTest.py',
name='Python Actor Test',
)
self.getPort('In').connect(self.pythonActor)
self.pythonActor.connect(self.getPort('Out'))
def trigger(self, inData):
self.getPort('In').trigger(inData)
class Workflow3(Workflow):
"""
Workflow containing one submodel.
"""
\ No newline at end of file
Workflow containing one start actor,
one submodel and one stop actor.
"""
def __init__(self, name):
Workflow.__init__(self, name)
self.startActor = StartActor()
self.submodel1 = Submodel1('Submodel 1')
self.stopActor = StopActor()
self.startActor.connect(self.submodel1.getPort('In'))
self.submodel1.getPort('Out').connect(self.stopActor)
def run(self, inData):
self.startActor.trigger(inData)
self.stopActor.join(timeout=5)
return self.stopActor.outData
class Submodel2(Submodel):
"""
Submodel containing one python actor which throws an exception.
"""
def __init__(self, name):
Submodel.__init__(self, name=name)
self.pythonActor = PythonActor(
script='pypushflow.test.pythonErrorHandlerTest.py',
name='Python Error Handler Test',
errorHandler=self
)
self.getPort('In').connect(self.pythonActor)
self.pythonActor.connect(self.getPort('Out'))
class Workflow4(Workflow):
"""
Workflow containing one start actor,
one submodel which throws an exception and one stop actor.
"""
def __init__(self, name):
Workflow.__init__(self, name)
self.startActor = StartActor()
self.submodel2 = Submodel2('Submodel 2')
self.stopActor = StopActor()
self.startActor.connect(self.submodel2.getPort('In'))
self.submodel2.getPort('Out').connect(self.stopActor)
def run(self, inData):
self.startActor.trigger(inData)
self.stopActor.join(timeout=5)
return self.stopActor.outData
......@@ -28,6 +28,7 @@ import unittest
from pypushflow.test.TestWorkflow import Workflow1
from pypushflow.test.TestWorkflow import Workflow2
from pypushflow.test.TestWorkflow import Workflow3
logging.basicConfig(level=logging.DEBUG)
......@@ -37,15 +38,30 @@ logger = logging.getLogger('testWorkflow')
class TestWorkflows(unittest.TestCase):
def test_Workflow1(self):
testWorkflow = Workflow1('Test workflow 1')
testWorkflow1 = Workflow1('Test workflow 1')
inData = {'name': 'Jerry'}
outData = testWorkflow.run(inData)
outData = testWorkflow1.run(inData)
self.assertIsNotNone(outData)
self.assertEqual(outData['reply'], 'Hello Jerry!')
def test_Workflow2(self):
testWorkflow = Workflow2('Test workflow 2')
inData = {'name': 'Jerry'}
outData = testWorkflow.run(inData)
testWorkflow2 = Workflow2('Test workflow 2')
inData = {'name': 'Tom'}
outData = testWorkflow2.run(inData)
self.assertIsNotNone(outData)
self.assertTrue('WorkflowException' in outData)
def test_Workflow3(self):
testWorkflow3 = Workflow3('Test workflow 3')
inData = {'name': 'Cat'}
outData = testWorkflow3.run(inData)
self.assertIsNotNone(outData)
self.assertEqual(outData['reply'], 'Hello Cat!')
def test_Workflow4(self):
testWorkflow4 = Workflow2('Test workflow 4')
inData = {'name': 'Dog'}
outData = testWorkflow4.run(inData)
self.assertIsNotNone(outData)
self.assertTrue('WorkflowException' in outData)
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment