%{
#include "KernelSDS.hxx"
+#include "Utils_SALOME_Exception.hxx"
%}
+
+namespace SALOME
+{
+ class SALOME_Exception
+ {
+ public:
+ %extend
+ {
+ std::string __str__() const
+ {
+ return std::string(self->what());
+ }
+ }
+ };
+}
+
+using namespace SALOME;
+
+%exception {
+ try {
+ $action
+ }
+ catch (SALOME::SALOME_Exception& _e) {
+ // Reraise with SWIG_Python_Raise
+ SWIG_Python_Raise(SWIG_NewPointerObj((new SALOME::SALOME_Exception(static_cast< const SALOME::SALOME_Exception& >(_e))),SWIGTYPE_p_INTERP_KERNEL__Exception,SWIG_POINTER_OWN), "INTERP_KERNEL::Exception", SWIGTYPE_p_INTERP_KERNEL__Exception);
+ SWIG_fail;
+ }
+}
+
%inline
{
- std::string GetDSMInstanceInternal();
+ std::string GetDSMInstanceInternal(PyObject *argv)
+ {
+ if(!PyList_Check(argv))
+ THROW_SALOME_EXCEPTION("Not a pylist");
+ Py_ssize_t sz=PyList_Size(pyLi);
+ std::vector<std::string> argvCpp(sz);
+ for(Py_ssize_t i = 0 ; i < sz ; ++i)
+ {
+ PyObject *obj = PyList_GetItem(pyLi,i);
+ if(!PyUnicode_Check(obj))
+ THROW_SALOME_EXCEPTION("Not a pylist of strings");
+ {
+ Py_ssize_t dummy;
+ argvCpp[i] = PyUnicode_AsUTF8AndSize(obj,&dummy);
+ }
+ }
+ GetDSMInstanceInternal(argvCpp);
+ }
}
%pythoncode %{
-def GetDSMInstance():
+def GetDSMInstance(argv):
import SALOME
import CORBA
orb=CORBA.ORB_init([''])
- return orb.string_to_object(GetDSMInstanceInternal())
+ return orb.string_to_object(GetDSMInstanceInternal(argv))
%}
time.sleep(3)
dss.atomicApply([t])
def work(t):
- i,varName,scopeName=t
+ IORNS,i,varName,scopeName=t
if i==0:
generateKey(varName,scopeName)
return 0
import TestSalomeSDSHelper0
import os,subprocess
fname=os.path.splitext(TestSalomeSDSHelper0.__file__)[0]+".py"
- proc = subprocess.Popen(["python3", fname], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+ proc = subprocess.Popen(["python3", fname, IORNS], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out,err=proc.communicate()
if proc.returncode!=0:
print("-------------- work -----------")
class SalomeSDSTest(unittest.TestCase):
- def testList1(self):
+ def tessList1(self):
a=SalomeSDSClt.CreateRdExtGlobalVar([],"a","Scope0")
self.assertEqual(a.local_copy(),[])
a.append(5)
a.ptr().getMyDataScopeServer().deleteVar("a")
pass
- def testDict1(self):
+ def tessDict1(self):
a=SalomeSDSClt.CreateRdExtGlobalVar({},"a","Scope0")
a["ab"]=4
self.assertEqual(a.local_copy(),{"ab":4})
a.ptr().getMyDataScopeServer().deleteVar("a")
pass
- def testReadOnly1(self):
+ def tessReadOnly1(self):
a=SalomeSDSClt.CreateRdOnlyGlobalVar({"ab":4,"cd":[5,77]},"a","Scope0")
self.assertEqual(a.local_copy(),{"ab":4,"cd":[5,77]})
self.assertRaises(Exception,a.__getitem__,"ab")
#
nbProc=8
pool=mp.Pool(processes=nbProc)
- asyncResult=pool.map_async(work,[(i,varName,scopeName) for i in range(nbProc)])
+ from NamingService import NamingService
+ asyncResult=pool.map_async(work,[(NamingService.IOROfNS(),i,varName,scopeName) for i in range(nbProc)])
print("asyncResult=", asyncResult)
self.assertEqual(asyncResult.get(),nbProc*[0]) # <- the big test is here !
dsm.removeDataScope(scopeName)
- def testTransaction2(self):
+ def tessTransaction2(self):
scopeName="Scope1"
varName="a"
dsm=salome.naming_service.Resolve("/DataServerManager")
wk.waitFor()
self.assertEqual(str2Obj(dss.waitForMonoThrRev(wk)),[7,8,9,10])
- def testTransaction3(self):
+ def tessTransaction3(self):
scopeName="Scope1"
varName="a"
dsm=salome.naming_service.Resolve("/DataServerManager")
dss.atomicApply([t2])
self.assertEqual(str2Obj(dss.fetchSerializedContent(varName)),{'cd':[7,8,9,10]})
- def testTransaction4(self):
+ def tessTransaction4(self):
scopeName="Scope1"
varName="a"
dsm=salome.naming_service.Resolve("/DataServerManager")
dss.atomicApply([t2])
self.assertEqual(str2Obj(dss.fetchSerializedContent(varName)),{'ab':[4,5,6]})
- def testTransaction5(self):
+ def tessTransaction5(self):
""" Like testTransaction2 but without transactions. """
scopeName="Scope1"
varName="a"
keys=[str2Obj(elt) for elt in dss.getAllKeysOfVarWithTypeDict(varName)]
self.assertEqual(set(keys),set(['ab','cd']))
- def testTransaction6(self):
+ def tessTransaction6(self):
""" Test to test RdWr global vars with transaction"""
scopeName="Scope1"
varName="a"
dsm.removeDataScope(scopeName)
pass
- def testTransaction7(self):
+ def tessTransaction7(self):
"""Like testTransaction5 but after a recovery."""
scopeName="Scope1"
varName="a"
self.assertEqual(str2Obj(dss.fetchSerializedContent(varName)),{'ab':[4,5,6],'cd':[7,8,9,10]})
pass
- def testTransaction8(self):
+ def tessTransaction8(self):
""" EDF 16833 and EDF17719 """
funcContent="""def comptchev(a,b):
return "d" not in a
self.assertEqual(str2Obj(dss.fetchSerializedContent(varName)),value3)
pass
- def testTransaction9(self):
+ def tessTransaction9(self):
""" EDF 16833 and EDF17719 : use case 2. Trying to createRdExt during add key session"""
funcContent="""def comptchev(a,b):
return a==b
pass
- def testLockToDump(self):
+ def tessLockToDump(self):
""" Test to check that holdRequests method. This method wait for clean server status and hold it until activeRequests is called.
Warning this method expects a not overloaded machine to be run because test is based on ellapse time.
"""
pass
def setUp(self):
- salome.salome_init()
+ #salome.salome_init()
+ salome.salome_init_without_session()
pass
pass
