-
Notifications
You must be signed in to change notification settings - Fork 3
/
test_processesbase.py
81 lines (64 loc) · 2.73 KB
/
test_processesbase.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
# -*- coding: utf-8 -*-
"""
Test provided base processes
@note: this file is here containing all default prcess tests one gets nasty
dependency probles with the python import system as we always need workflows for testing
"""
import unittest
import workflows
from mayarv.automation.report import Plan
import processes
import tempfile
class TestProcesses( unittest.TestCase ):
"""Test workflow class"""
def test_workflowProcess( self ):
wfl = workflows.workflowwrap
workflows.multiinput.writeDot(tempfile.gettempdir() + "/mygraph.dot" )
self.failUnless( len( list( wfl.iterNodes() ) ) == 1 )
rate, process = wfl.getTargetRating( unicode( "this" ) )
self.failUnless( rate != 0 )
# shuold be able to provide exactly the same output the workflow itself
res = wfl.makeTarget( unicode( "this" ) )
print res
self.failUnless( res == "this10.020202020202020202020202020202020" )
# CALLGRAPH
################
# nested nodes are containers that do not show as they do not do anything
# they are just containers after all and should yield the same result if
# compared to the unwrapped workflow
miwfl = workflows.multiinput
miwfl.makeTarget( unicode( "this" ) )
self.failUnless( miwfl._callgraph.number_of_nodes() == wfl._callgraph.number_of_nodes() )
# NESTED WFLS AND PLANS
########################
#print wfl._callgraph.nodes()
plan = wfl.getReportInstance( Plan )
lines = plan.getReport( headline = "WRAPPED WORKFLOW" )
# MULTI-NESTED WORKFLOW
########################
# Two wrapped workflows combined
mwfl = workflows.multiWorkflow
# iterate it - nodes should be facaded and you should not get inside
# we cannot get different nodes than workflow wrappers, even if we
# traverse the connections
wnode2 = mwfl.getNodes()[-1]
lastshell = None
for shell in wnode2.outChain.iterShells( direction = "up" ):
self.failUnless( isinstance( shell.node, processes.WorkflowWrapTestProcess ) )
lastshell = shell
for shell in lastshell.iterShells( direction = "down" ):
self.failUnless( isinstance( shell.node, processes.WorkflowWrapTestProcess ) )
res = mwfl.makeTarget( list( (5,) ) )[0] # target only
self.failUnless( res == 65 ) # it went through 6 nodes
# compute the value directly using the plug as the workflow
# cannot determine which input is the suitable one
#self.failUnless( miwfl.makeTarget( object )[0] < res )
self.failUnlessRaises( AssertionError, miwfl.makeTarget, object )
mires = miwfl.getNodes()[-1].outChain.get( )
self.failUnless( mires[0] < res ) # must be 3 nodes only, thus its smaller at least
# report
plan = mwfl.getReportInstance( Plan )
lines = plan.getReport( headline = "MULTI WRAPPED WORKFLOW" )
self.failUnless( len( lines ) == 7 )
for l in lines:
print l