src/yacsloader_swig/Test/testValidationChecks.py

   1 #! /usr/bin/env python3
   2 # -*- coding: utf-8 -*-
   3 # Copyright (C) 2006-2022  CEA/DEN, EDF R&D
   4 #
   5 # This library is free software; you can redistribute it and/or
   6 # modify it under the terms of the GNU Lesser General Public
   7 # License as published by the Free Software Foundation; either
   8 # version 2.1 of the License, or (at your option) any later version.
   9 #
  10 # This library is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13 # Lesser General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU Lesser General Public
  16 # License along with this library; if not, write to the Free Software
  17 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  18 #
  19 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  20 #
  21
  22 import unittest
  23 import os
  24 import sys
  25 import pilot
  26 import SALOMERuntime
  27
  28 class TestValidationChecks(unittest.TestCase):
  29   def test_foreach_links(self):
  30     """ Test tha add of illegal links within a foreach loop.
  31     """
  32     SALOMERuntime.RuntimeSALOME_setRuntime()
  33     runtime = pilot.getRuntime()
  34
  35     schema = runtime.createProc("schema")
  36     ti=schema.getTypeCode("int")
  37     tiset = schema.createSequenceTc("", "seqint", ti)
  38
  39     nw = runtime.createScriptNode("", "Worker")
  40     nw.edAddInputPort("i1", ti)
  41     nw.edAddInputPort("i2", ti)
  42     nw.edAddOutputPort("o1", ti)
  43     nw.setScript("o1=i1+i2")
  44
  45     ni = runtime.createScriptNode("", "Init")
  46     ni.edAddInputPort("ii", ti)
  47     ni.edAddOutputPort("oi", ti)
  48     ni.setScript("o1=ii")
  49
  50     nf = runtime.createScriptNode("", "Fin")
  51     nf.edAddInputPort("ifin", ti)
  52     nf.edAddOutputPort("ofin", ti)
  53     nf.setScript("ofin=ifin")
  54
  55     npre = runtime.createScriptNode("", "PreProc")
  56     npre.edAddOutputPort("opre", ti)
  57     npre.setScript("opre=5")
  58
  59     npost = runtime.createScriptNode("", "PostProc")
  60     npost.edAddInputPort("ipost", ti)
  61
  62     fe = runtime.createForEachLoop("ForEach", ti)
  63     fe.getInputPort("nbBranches").edInitPy(2)
  64     fe.getInputPort("SmplsCollection").edInitPy([1, 2, 3, 4])
  65     fe.edSetNode(nw)
  66     fe.edSetInitNode(ni)
  67     fe.edSetFinalizeNode(nf)
  68
  69     schema.edAddChild(fe)
  70     schema.edAddChild(npost)
  71
  72     def excpTest(op, ip):
  73       self.assertRaises(ValueError, schema.edAddLink, op, ip)
  74       pass
  75
  76     # ForEachLoop tests
  77     excpTest(fe.getOutputPort("evalSamples"), npost.getInputPort("ipost"))
  78     excpTest(fe.getOutputPort("evalSamples"), ni.getInputPort("ii"))
  79     excpTest(fe.getOutputPort("evalSamples"), nf.getInputPort("ifin"))
  80     excpTest(fe.getOutputPort("evalSamples"), fe.getInputPort("nbBranches"))
  81     excpTest(ni.getOutputPort("oi"), fe.getInputPort("nbBranches"))
  82     excpTest(nw.getOutputPort("o1"), fe.getInputPort("nbBranches"))
  83     excpTest(nf.getOutputPort("ofin"), fe.getInputPort("nbBranches"))
  84     excpTest(fe.getOutputPort("evalSamples"), fe.getInputPort("SmplsCollection"))
  85     excpTest(ni.getOutputPort("oi"), fe.getInputPort("SmplsCollection"))
  86     excpTest(nw.getOutputPort("o1"), fe.getInputPort("SmplsCollection"))
  87     excpTest(nf.getOutputPort("ofin"), fe.getInputPort("SmplsCollection"))
  88     excpTest(nw.getOutputPort("o1"), nf.getInputPort("ifin"))
  89     excpTest(nw.getOutputPort("o1"), ni.getInputPort("ii"))
  90     excpTest(ni.getOutputPort("oi"), npost.getInputPort("ipost"))
  91     excpTest(nf.getOutputPort("ofin"), npost.getInputPort("ipost"))
  92     excpTest(nf.getOutputPort("ofin"), ni.getInputPort("ii"))
  93
  94   def test_optim_links(self):
  95     """ Test tha add of illegal links within an optimization loop.
  96     """
  97     SALOMERuntime.RuntimeSALOME_setRuntime()
  98     runtime = pilot.getRuntime()
  99     #
 100     schema = runtime.createProc("schema")
 101     ti=schema.getTypeCode("int")
 102     #
 103     nw = runtime.createScriptNode("", "Worker")
 104     nw.edAddInputPort("i1", ti)
 105     nw.edAddInputPort("i2", ti)
 106     nw.edAddOutputPort("o1", ti)
 107     nw.setScript("o1=i1+i2")
 108     #
 109     ni = runtime.createScriptNode("", "Init")
 110     ni.edAddInputPort("ii", ti)
 111     ni.edAddOutputPort("oi", ti)
 112     ni.setScript("o1=ii")
 113     #
 114     nf = runtime.createScriptNode("", "Fin")
 115     nf.edAddInputPort("ifin", ti)
 116     nf.edAddOutputPort("ofin", ti)
 117     nf.setScript("ofin=ifin")
 118     #
 119     npre = runtime.createScriptNode("", "PreProc")
 120     npre.edAddOutputPort("opre", ti)
 121     npre.setScript("opre=5")
 122     #
 123     npost = runtime.createScriptNode("", "PostProc")
 124     npost.edAddInputPort("ipost", ti)
 125     #
 126     fe = runtime.createOptimizerLoop("OptLoop", "optim_plugin.py","myalgosync",True)
 127     fe.getInputPort("nbBranches").edInitPy(2)
 128     fe.getInputPort("algoInit").edInitPy(7)
 129     fe.edSetNode(nw)
 130     fe.edSetInitNode(ni)
 131     fe.edSetFinalizeNode(nf)
 132     #
 133     schema.edAddChild(fe)
 134     schema.edAddChild(npost)
 135     schema.edAddChild(npre)
 136
 137     def excpTest(op, ip):
 138       self.assertRaises(ValueError, schema.edAddLink, op, ip)
 139       pass
 140
 141     # ForEachLoop tests
 142     excpTest(fe.getOutputPort("evalSamples"), npost.getInputPort("ipost"))
 143     excpTest(fe.getOutputPort("evalSamples"), ni.getInputPort("ii"))
 144     excpTest(fe.getOutputPort("evalSamples"), nf.getInputPort("ifin"))
 145     excpTest(fe.getOutputPort("evalSamples"), fe.getInputPort("nbBranches"))
 146     excpTest(nw.getOutputPort("o1"), nf.getInputPort("ifin"))
 147     excpTest(nw.getOutputPort("o1"), ni.getInputPort("ii"))
 148     excpTest(ni.getOutputPort("oi"), npost.getInputPort("ipost"))
 149     excpTest(nf.getOutputPort("ofin"), npost.getInputPort("ipost"))
 150     excpTest(nf.getOutputPort("ofin"), ni.getInputPort("ii"))
 151     # Specific OptimizerLoop tests
 152     excpTest(nw.getOutputPort("o1"), fe.getInputPort("nbBranches"))
 153     excpTest(ni.getOutputPort("oi"), fe.getInputPort("nbBranches"))
 154     excpTest(nf.getOutputPort("ofin"), fe.getInputPort("nbBranches"))
 155     excpTest(nw.getOutputPort("o1"), fe.getInputPort("algoInit"))
 156     excpTest(ni.getOutputPort("oi"), fe.getInputPort("algoInit"))
 157     excpTest(nf.getOutputPort("ofin"), fe.getInputPort("algoInit"))
 158     excpTest(ni.getOutputPort("oi"), fe.getInputPort("evalResults"))
 159     excpTest(nf.getOutputPort("ofin"), fe.getInputPort("evalResults"))
 160     excpTest(npre.getOutputPort("opre"), fe.getInputPort("evalResults"))
 161     excpTest(fe.getOutputPort("evalSamples"), fe.getInputPort("algoInit"))
 162     excpTest(fe.getOutputPort("algoResults"), nw.getInputPort("i1"))
 163     excpTest(fe.getOutputPort("algoResults"), ni.getInputPort("ii"))
 164     excpTest(fe.getOutputPort("algoResults"), nf.getInputPort("ifin"))
 165     excpTest(nw.getOutputPort("o1"), npost.getInputPort("ipost"))
 166     excpTest(ni.getOutputPort("oi"), npost.getInputPort("ipost"))
 167     excpTest(nf.getOutputPort("ofin"), npost.getInputPort("ipost"))
 168
 169 if __name__ == '__main__':
 170     unittest.main()