src/yacsloader_swig/Test/StdAloneYacsLoaderTest1.py

   1 # Copyright (C) 2006-2015  CEA/DEN, EDF R&D
   2 #
   3 # This library is free software; you can redistribute it and/or
   4 # modify it under the terms of the GNU Lesser General Public
   5 # License as published by the Free Software Foundation; either
   6 # version 2.1 of the License, or (at your option) any later version.
   7 #
   8 # This library is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11 # Lesser General Public License for more details.
  12 #
  13 # You should have received a copy of the GNU Lesser General Public
  14 # License along with this library; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  16 #
  17 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  18 #
  19
  20 import pilot
  21 import SALOMERuntime
  22 import loader
  23 import unittest
  24
  25 class StdAloneYacsLoaderTest1(unittest.TestCase):
  26
  27   def setUp(self):
  28     SALOMERuntime.RuntimeSALOME_setRuntime()
  29     self.r = pilot.getRuntime()
  30     self.l = loader.YACSLoader()# self.l.load("foreachImbr_tmp.xml")
  31     pass
  32
  33   def test1(self):
  34     """tests imbrication of foreach loop."""
  35     SALOMERuntime.RuntimeSALOME_setRuntime()
  36     l=loader.YACSLoader()
  37     ex=pilot.ExecutorSwig()
  38     p=self.r.createProc("pr")
  39     td=p.createType("double","double")
  40     td2=p.createSequenceTc("seqdbl","seqdbl",td)
  41     td3=p.createSequenceTc("seqdblvec","seqdblvec",td2)
  42     td4=p.createSequenceTc("seqseqdblvec","seqseqdblvec",td3)
  43     node1=self.r.createScriptNode("","node1")
  44     node1.setScript("o1=[([1,1],[2,2,2]),([10],[11,11],[12,12,12]),([20],[21,21],[22,22,22],[23,23,23,23])]")
  45     o1=node1.edAddOutputPort("o1",td4)
  46     p.edAddChild(node1)
  47     node2=self.r.createForEachLoop("node2",td3)
  48     p.edAddChild(node2)
  49     p.edAddCFLink(node1,node2)
  50     p.edAddLink(o1,node2.edGetSeqOfSamplesPort())
  51     node2.edGetNbOfBranchesPort().edInitInt(2)
  52     #
  53     node20=self.r.createBloc("node20")
  54     node2.edAddChild(node20)
  55     node200=self.r.createForEachLoop("node200",td2)
  56     node20.edAddChild(node200)
  57     node200.edGetNbOfBranchesPort().edInitInt(2)
  58     p.edAddLink(node2.edGetSamplePort(),node200.edGetSeqOfSamplesPort())
  59     node2000=self.r.createForEachLoop("node2000",td)
  60     node2000.edGetNbOfBranchesPort().edInitInt(2)
  61     node200.edAddChild(node2000)
  62     p.edAddLink(node200.edGetSamplePort(),node2000.edGetSeqOfSamplesPort())
  63     node20000=self.r.createScriptNode("","node20000")
  64     node2000.edAddChild(node20000)
  65     i1=node20000.edAddInputPort("i1",td)
  66     o2=node20000.edAddOutputPort("o2",td)
  67     node20000.setScript("o2=i1+2")
  68     p.edAddLink(node2000.edGetSamplePort(),i1)
  69     #
  70     node3=self.r.createScriptNode("","node3")
  71     node3.setScript("o3=i2")
  72     p.edAddChild(node3)
  73     i2=node3.edAddInputPort("i2",td4)
  74     o3=node3.edAddOutputPort("o3",td4)
  75     p.edAddCFLink(node2,node3)
  76     p.edAddLink(o2,i2)
  77     ex = pilot.ExecutorSwig()
  78     self.assertEqual(p.getState(),pilot.READY)
  79     ex.RunW(p,0)
  80     self.assertEqual(p.getState(),pilot.DONE)
  81     zeResu=node3.getOutputPort("o3").get()
  82     self.assertEqual(zeResu,[[[3.,3.],[4.,4.,4.]],[[12.],[13.,13.],[14.,14.,14.]],[[22.],[23.,23.],[24.,24.,24.],[25.,25.,25.,25.]]])
  83     p.saveSchema("foreachImbrBuildFS.xml")
  84     pass
  85
  86   def test2(self):
  87     """ Non regression test. When input/output declared as pyobj hiding a string type to go to or from a ForEachLoop it previous lead
  88     to an error.
  89     """
  90     fname="BugPyObjStrInYacs.xml"
  91     p=self.r.createProc("pr")
  92     tc0=p.createInterfaceTc("python:obj:1.0","pyobj",[])
  93     tc1=p.createSequenceTc("list[pyobj]","list[pyobj]",tc0)
  94     #
  95     node0=self.r.createScriptNode("Salome","node0")
  96     node0.setScript("o1=[\"a\",\"bc\"]")
  97     o1=node0.edAddOutputPort("o1",tc1)
  98     p.edAddChild(node0)
  99     #
 100     node1=self.r.createForEachLoop("node1",tc0)
 101     p.edAddChild(node1)
 102     p.edAddCFLink(node0,node1)
 103     node1.edGetNbOfBranchesPort().edInitInt(1)
 104     p.edAddLink(o1,node1.edGetSeqOfSamplesPort())
 105     #
 106     node10=self.r.createScriptNode("Salome","node10")
 107     node10.setScript("o1=3*i1")
 108     i10_1=node10.edAddInputPort("i1",tc0)
 109     o10_1=node10.edAddOutputPort("o1",tc0)
 110     node1.edAddChild(node10)
 111     p.edAddLink(node1.edGetSamplePort(),i10_1)
 112     #
 113     node2=self.r.createScriptNode("Salome","node2")
 114     node2.setScript("o1=i1")
 115     i2_1=node2.edAddInputPort("i1",tc1)
 116     o2_1=node2.edAddOutputPort("o1",tc1)
 117     p.edAddChild(node2)
 118     p.edAddCFLink(node1,node2)
 119     p.edAddLink(o10_1,i2_1)
 120     ##
 121     p.saveSchema(fname)
 122     p=self.l.load(fname)
 123     ##
 124     ex = pilot.ExecutorSwig()
 125     self.assertEqual(p.getState(),pilot.READY)
 126     ex.RunW(p,0)
 127     self.assertEqual(p.getState(),pilot.DONE)
 128     #
 129     self.assertEqual(p.getChildByName("node2").getOutputPort("o1").get(),['aaa','bcbcbc'])
 130     pass
 131
 132   def test3(self):
 133     fname="BugInConcurrentLaunchDftCont.xml"
 134     p=self.r.createProc("pr")
 135     ti=p.createType("int","int")
 136     cont=p.createContainer("DefaultContainer","Salome")
 137     cont.setProperty("container_name","FactoryServer")
 138     b=self.r.createBloc("Bloc") ; p.edAddChild(b)
 139     #
 140     nb=4
 141     outs=[]
 142     for i in xrange(nb):
 143       node=self.r.createScriptNode("Salome","node%d"%i)
 144       node.setExecutionMode("remote")
 145       node.setContainer(cont)
 146       outs.append(node.edAddOutputPort("i",ti))
 147       node.setScript("i=%d"%i)
 148       b.edAddChild(node)
 149     #
 150     node=self.r.createScriptNode("Salome","nodeEnd")
 151     node.setExecutionMode("remote")
 152     node.setContainer(cont)
 153     res=node.edAddOutputPort("res",ti)
 154     p.edAddChild(node)
 155     l=[]
 156     for i in xrange(nb):
 157       elt="i%d"%i
 158       inp=node.edAddInputPort(elt,ti) ; l.append(elt)
 159       p.edAddChild(node)
 160       p.edAddLink(outs[i],inp)
 161     node.setScript("res="+"+".join(l))
 162     p.edAddCFLink(b,node)
 163     #
 164     for i in xrange(10):
 165       p.init()
 166       ex = pilot.ExecutorSwig()
 167       self.assertEqual(p.getState(),pilot.READY)
 168       ex.RunW(p,0)
 169       self.assertEqual(res.get(),6)
 170       self.assertEqual(p.getState(),pilot.DONE)
 171     pass
 172
 173   def tearDown(self):
 174     del self.r
 175     del self.l
 176     pass
 177
 178   pass
 179
 180 if __name__ == '__main__':
 181     unittest.main()