[openrtm-commit:01859] r698 - trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test
openrtm @ openrtm.org
openrtm @ openrtm.org
2016年 3月 16日 (水) 15:49:01 JST
Author: miyamoto
Date: 2016-03-16 15:49:01 +0900 (Wed, 16 Mar 2016)
New Revision: 698
Added:
trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTransport.py
Removed:
trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTrasport.py
Log:
[compat,bugfix,->RELENG_1_2] bug fix.
Added: trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTransport.py
===================================================================
--- trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTransport.py (rev 0)
+++ trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTransport.py 2016-03-16 06:49:01 UTC (rev 698)
@@ -0,0 +1,129 @@
+#!/usr/bin/env python
+# -*- coding: euc-jp -*-
+
+#
+# \file test_SSLTrasport.py
+# \brief
+# \date $Date: $
+# \author Nobuhiko Miyamoto
+#
+
+
+import sys
+sys.path.insert(1,"../")
+
+try:
+ import unittest2 as unittest
+except (ImportError):
+ import unittest
+
+import time
+
+#from Manager import *
+import OpenRTM_aist
+import RTC
+import multiprocessing
+import os
+import threading
+
+
+
+def RunOutPort(q):
+
+ argv = sys.argv[:]
+ #argv.extend(['-o', 'corba.endpoint::2810'])
+
+ manager = OpenRTM_aist.Manager.init(argv)
+ manager.activateManager()
+ _d_out = RTC.TimedLong(RTC.Time(0,0),0)
+ _outOut = OpenRTM_aist.OutPort("out", _d_out)
+ prop = OpenRTM_aist.Properties()
+ _outOut.init(prop)
+
+
+
+ """orb = manager.getORB()
+ poa = orb.resolve_initial_references("omniINSPOA")
+ poa._get_the_POAManager().activate()
+ id = "test"
+ poa.activate_object_with_id(id, _outOut)"""
+
+ manager._namingManager.bindPortObject("test", _outOut)
+
+
+ q.get()
+
+ _d_out.data = 100
+ _outOut.write()
+
+ q.get()
+
+
+
+ manager.shutdown()
+
+
+class TestSSLTransport(unittest.TestCase):
+
+ def setUp(self):
+ self.queue = multiprocessing.Queue()
+
+ sys.argv.extend(['-o', 'manager.preload.modules:SSLTransport.py'])
+ sys.argv.extend(['-o', 'manager.modules.load_path:./,../'])
+ sys.argv.extend(['-o', 'corba.ssl.certificate_authority_file:root.crt'])
+ sys.argv.extend(['-o', 'corba.ssl.key_file:server.pem'])
+ sys.argv.extend(['-o', 'corba.ssl.key_file_password:password'])
+ os.environ['ORBsslVerifyMode'] = "none"
+ self.outport_process = multiprocessing.Process(target=RunOutPort, args=(self.queue,))
+ self.outport_process.start()
+
+
+ time.sleep(1)
+ os.environ['ORBtraceLevel'] = '25'
+
+
+ self.manager = OpenRTM_aist.Manager.init(sys.argv)
+ self.manager.activateManager()
+ self._d_in = RTC.TimedLong(RTC.Time(0,0),0)
+ self._inIn = OpenRTM_aist.InPort("in", self._d_in)
+ prop = OpenRTM_aist.Properties()
+ self._inIn.init(prop)
+ self.inport_obj = self._inIn.getPortRef()
+
+
+
+ orb = self.manager.getORB()
+ #outport_name = "corbaloc:iiop:localhost:2810/test"
+ outport_name = "corbaname::localhost:2809/NameService#test"
+ oobj = orb.string_to_object(outport_name)
+ self.outport_obj = oobj._narrow(RTC.PortService)
+
+
+
+ def tearDown(self):
+ self.manager.shutdownManager()
+ self.queue.put("")
+
+ def test_Connect(self):
+
+ prop = OpenRTM_aist.Properties()
+ ret = OpenRTM_aist.connect("con1",prop,self.inport_obj,self.outport_obj)
+
+ self.queue.put("")
+
+ #ret = self._inIn.isNew()
+ #data = self._inIn.read()
+
+
+
+ #self.outport_obj.disconnect_all()
+
+
+
+
+
+
+
+############### test #################
+if __name__ == '__main__':
+ unittest.main()
Deleted: trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTrasport.py
===================================================================
--- trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTrasport.py 2016-03-16 05:29:41 UTC (rev 697)
+++ trunk/OpenRTM-aist-Python/OpenRTM_aist/ext/ssl/test/test_SSLTrasport.py 2016-03-16 06:49:01 UTC (rev 698)
@@ -1,129 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: euc-jp -*-
-
-#
-# \file test_SSLTrasport.py
-# \brief
-# \date $Date: $
-# \author Nobuhiko Miyamoto
-#
-
-
-import sys
-sys.path.insert(1,"../")
-
-try:
- import unittest2 as unittest
-except (ImportError):
- import unittest
-
-import time
-
-#from Manager import *
-import OpenRTM_aist
-import RTC
-import multiprocessing
-import os
-import threading
-
-
-
-def RunOutPort(q):
-
- argv = sys.argv[:]
- #argv.extend(['-o', 'corba.endpoint::2810'])
-
- manager = OpenRTM_aist.Manager.init(argv)
- manager.activateManager()
- _d_out = RTC.TimedLong(RTC.Time(0,0),0)
- _outOut = OpenRTM_aist.OutPort("out", _d_out)
- prop = OpenRTM_aist.Properties()
- _outOut.init(prop)
-
-
-
- """orb = manager.getORB()
- poa = orb.resolve_initial_references("omniINSPOA")
- poa._get_the_POAManager().activate()
- id = "test"
- poa.activate_object_with_id(id, _outOut)"""
-
- manager._namingManager.bindPortObject("test", _outOut)
-
-
- q.get()
-
- _d_out.data = 100
- _outOut.write()
-
- q.get()
-
-
-
- manager.shutdownManager()
-
-
-class TestSSLTransport(unittest.TestCase):
-
- def setUp(self):
- self.queue = multiprocessing.Queue()
-
- sys.argv.extend(['-o', 'manager.preload.modules:SSLTransport.py'])
- sys.argv.extend(['-o', 'manager.modules.load_path:./,../'])
- sys.argv.extend(['-o', 'corba.ssl.certificate_authority_file:root.crt'])
- sys.argv.extend(['-o', 'corba.ssl.key_file:server.pem'])
- sys.argv.extend(['-o', 'corba.ssl.key_file_password:password'])
- os.environ['ORBsslVerifyMode'] = "none"
- self.outport_process = multiprocessing.Process(target=RunOutPort, args=(self.queue,))
- self.outport_process.start()
-
-
- time.sleep(1)
- os.environ['ORBtraceLevel'] = '25'
-
-
- self.manager = OpenRTM_aist.Manager.init(sys.argv)
- self.manager.activateManager()
- self._d_in = RTC.TimedLong(RTC.Time(0,0),0)
- self._inIn = OpenRTM_aist.InPort("in", self._d_in)
- prop = OpenRTM_aist.Properties()
- self._inIn.init(prop)
- self.inport_obj = self._inIn.getPortRef()
-
-
-
- orb = self.manager.getORB()
- #outport_name = "corbaloc:iiop:localhost:2810/test"
- outport_name = "corbaname::localhost:2809/NameService#test"
- oobj = orb.string_to_object(outport_name)
- self.outport_obj = oobj._narrow(RTC.PortService)
-
-
-
- def tearDown(self):
- self.manager.shutdownManager()
- self.queue.put("")
-
- def test_Connect(self):
-
- prop = OpenRTM_aist.Properties()
- ret = OpenRTM_aist.connect("con1",prop,self.inport_obj,self.outport_obj)
-
- self.queue.put("")
-
- #ret = self._inIn.isNew()
- #data = self._inIn.read()
-
-
-
- #self.outport_obj.disconnect_all()
-
-
-
-
-
-
-
-############### test #################
-if __name__ == '__main__':
- unittest.main()
More information about the openrtm-commit
mailing list