2 # Copyright (C) 2006-2024 CEA, EDF
4 # This library is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU Lesser General Public
6 # License as published by the Free Software Foundation; either
7 # version 2.1 of the License, or (at your option) any later version.
9 # This library is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
12 # Lesser General Public License for more details.
14 # You should have received a copy of the GNU Lesser General Public
15 # License along with this library; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
29 from SALOME_PyNode import UnProxyObjectSimple
31 dir_test = tempfile.mkdtemp(suffix=".yacstest")
33 class TestEdit(unittest.TestCase):
36 SALOMERuntime.RuntimeSALOME_setRuntime()
37 self.r = pilot.getRuntime()
38 self.l = loader.YACSLoader()
39 self.e = pilot.ExecutorSwig()
44 cm = salome.lcc.getContainerManager()
45 cm.ShutdownContainers()
48 """ Test the conservation of the python context between two nodes sharing
55 proc=runtime.createProc("MySchema")
56 ti=proc.createType("int","int")
57 cont=proc.createContainer("MyContainer","Salome")
58 # type "multi" : the workload manager chooses the resource
59 # type "mono" : the resource is chosen by kernel, using the old rules.
60 cont.setProperty("type","multi")
61 # number of cores used by the container
62 cont.setProperty("nb_parallel_procs", "1")
63 n1=runtime.createScriptNode("","n1")
64 n2=runtime.createScriptNode("","n2")
65 n1.setExecutionMode("remote")
66 n2.setExecutionMode("remote")
70 res_port=n2.edAddOutputPort("v", ti)
73 proc.edAddCFLink(n1,n2)
74 # set the default execution mode using the workload manager
75 # if no property is set, the old executor is used
76 proc.setProperty("executor", "workloadmanager")
77 # reuse the same python context for every execution
78 cont.usePythonCache(True)
80 schema_file = os.path.join(dir_test,"pynode_with_cache1.xml")
81 proc.saveSchema(schema_file)
82 reloaded_proc = yacsloader.load(schema_file)
83 # default run method of the executor which uses the property "executor"
84 # in order to choose the actual run method
85 executor.RunW(reloaded_proc,0)
86 # you can also directly call the executor you wish, ignoring the property
87 #executor.RunB(proc,0) # always use the "old" executor
88 #executor.runWlm(proc,0) # always use the workload manager based executor
89 reloaded_res_port = reloaded_proc.getChildByName("n2").getOutputPort("v")
90 self.assertEqual(UnProxyObjectSimple( reloaded_res_port.getPyObj() ), 42)
93 """ Same as test1, but using the old executor instead of workload manager.
98 proc=runtime.createProc("MySchema")
99 ti=proc.createType("int","int")
100 cont=proc.createContainer("MyContainer","Salome")
101 # With the old executor the type multi imposes the creation of a new
102 # container for every node. We need the type "mono" in order to have
103 # the same container used for both yacs nodes.
104 cont.setProperty("type","mono")
105 n1=runtime.createScriptNode("","n1")
106 n2=runtime.createScriptNode("","n2")
107 n1.setExecutionMode("remote")
108 n2.setExecutionMode("remote")
109 n1.setContainer(cont)
110 n2.setContainer(cont)
112 res_port=n2.edAddOutputPort("v", ti)
115 proc.edAddCFLink(n1,n2)
116 # reuse the same python context for every execution
117 cont.usePythonCache(True)
119 schema_file = os.path.join(dir_test,"pynode_with_cache2.xml")
120 proc.saveSchema(schema_file)
121 reloaded_proc = yacsloader.load(schema_file)
122 # default run method of the executor which uses the property "executor"
123 # in order to choose the actual run method
124 executor.RunW(reloaded_proc,0)
125 # you can also directly call the executor you wish, ignoring the property
126 #executor.RunB(proc,0) # always use the "old" executor
127 reloaded_res_port = reloaded_proc.getChildByName("n2").getOutputPort("v")
128 self.assertEqual(UnProxyObjectSimple( reloaded_res_port.getPyObj() ), 42)
130 if __name__ == '__main__':
131 file_test = os.path.join(dir_test,"UnitTestsResult")
132 with open(file_test, 'a') as f:
133 f.write(" --- TEST src/yacsloader: testPynodeWithCache.py\n")
134 suite = unittest.makeSuite(TestEdit)
135 result=unittest.TextTestRunner(f, descriptions=1, verbosity=1).run(suite)
136 sys.exit(not result.wasSuccessful())