test_PeriodicECSharedComposite.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- Python -*-
 
 
 
 
import sys
sys.path.insert(1,"../")
sys.path.insert(1,"../RTM_IDL")
 
import OpenRTM_aist
import RTC
 
import unittest
 
 
configsample_spec = ["implementation_id", "TestComp",
                     "type_name",         "TestComp",
                     "description",       "Test example component",
                     "version",           "1.0",
                     "vendor",            "Shinji Kurihara, AIST",
                     "category",          "example",
                     "activity_type",     "DataFlowComponent",
                     "max_instance",      "10",
                     "language",          "Python",
                     "lang_type",         "script",
                     # Configuration variables
                     "conf.default.int_param0", "0",
                     "conf.default.int_param1", "1",
                     "conf.default.double_param0", "0.11",
                     "conf.default.double_param1", "9.9",
                     "conf.default.str_param0", "hoge",
                     "conf.default.str_param1", "dara",
                     "conf.default.vector_param0", "0.0,1.0,2.0,3.0,4.0",
                     ""]
 
com = None
 
 
class TestComp(OpenRTM_aist.DataFlowComponentBase):
 
        def __init__(self, manager):
                OpenRTM_aist.DataFlowComponentBase.__init__(self, manager)
 
        def onInitialize(self):
                print("onInitialize")
                return RTC.RTC_OK
 
        def onFinalize(self):
                print("onFinalize")
                return RTC.RTC_OK
                
        def onStartup(self, ec_id):
                print("onStartup")
                return RTC.RTC_OK
 
        def onShutdown(self, ec_id):
                print("onSutdown")
                return RTC.RTC_OK
 
        def onActivated(self, ec_id):
                print("onActivated")
                return RTC.RTC_OK
 
        def onDeactivated(self, ec_id):
                print("onDeactivated")
                return RTC.RTC_OK
 
        def onExecute(self, ec_id):
                print("onExecute")
                return RTC.RTC_OK
 
        def onAborting(self, ec_id):
                print("onAborting")
                return RTC.RTC_OK
 
        def onReset(self, ec_id):
                print("onReset")
                return RTC.RTC_OK
                
        def onStateUpdate(self, ec_id):
                print("onStateUpdate")
                return RTC.RTC_OK
 
        def onRateChanged(self, ec_id):
                print("onRateChanged")
                return RTC.RTC_OK
 
                
def TestCompInit(manager):
        global com
        profile = OpenRTM_aist.Properties(defaults_str=configsample_spec)
        manager.registerFactory(profile,
                                TestComp,
                                OpenRTM_aist.Delete)
        com = manager.createComponent("TestComp")
 
 
class TestPeriodicECSharedComposite(unittest.TestCase):
 
        def setUp(self):
                self.manager = OpenRTM_aist.Manager.init(sys.argv)
                self.composite = OpenRTM_aist.PeriodicECSharedComposite(self.manager)
                return
 
        def test_organization(self):
                global com
                self.manager.setModuleInitProc(TestCompInit)
                self.manager.activateManager()
                organization = OpenRTM_aist.PeriodicECOrganization(self.composite)
                self.assertEqual(organization.set_members([com.getObjRef()]),True)
                #self.assertEqual(organization.add_members([com.getObjRef()]),True)
                self.assertEqual(organization.remove_member("TestComp0"),True)
                self.assertEqual(organization.add_members([com.getObjRef()]),True)
                organization.removeAllMembers()
                return
 
        
        def test_onInitialize(self):
                self.assertEqual(self.composite.onInitialize(), RTC.RTC_OK)
                return
        
        
        def test_onActivated(self):
                self.assertEqual(self.composite.onActivated(None), RTC.RTC_OK)
                return
        
        
        def test_onDeactivated(self):
                self.assertEqual(self.composite.onDeactivated(None), RTC.RTC_OK)
                return
        
        
        def test_onReset(self):
                self.assertEqual(self.composite.onReset(None), RTC.RTC_OK)
                return
 
 
        def test_delegatePort(self):
                return
 
 
 
 
if __name__ == '__main__':
        unittest.main()