00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016 import sys
00017 sys.path.insert(1, "../")
00018
00019 import unittest
00020 from SdoServiceAdmin import *
00021 import OpenRTM_aist
00022 import SDOPackage,SDOPackage__POA
00023 from omniORB import CORBA, PortableServer
00024 from omniORB import any
00025
00026
00027 class MockRTC(OpenRTM_aist.RTObject_impl):
00028 def __init__(self):
00029 self._orb = CORBA.ORB_init()
00030 self._poa = self._orb.resolve_initial_references("RootPOA")
00031 OpenRTM_aist.RTObject_impl.__init__(self, orb=self._orb, poa=self._poa)
00032 pass
00033
00034
00035 class SDOServiceMock(SDOPackage__POA.SDOService):
00036 def __init__(self):
00037 pass
00038
00039 class MockSdoServiceConsumer(OpenRTM_aist.SdoServiceConsumerBase):
00040 """
00041 """
00042
00043 def __init__(self):
00044 OpenRTM_aist.SdoServiceConsumerBase.__init__(self)
00045 return
00046
00047 def reinit(self, profile):
00048 pass
00049
00050 def getProfile(self):
00051 any_val = any.to_any("3.14159")
00052 nv = SDOPackage.NameValue("PROPERTIES NAME 0", any_val)
00053 sprof = SDOPackage.ServiceProfile("test id","INTERFACE_TYPE",[nv],SDOServiceMock())
00054 return sprof
00055
00056
00057 class TestListener(unittest.TestCase):
00058 def setUp(self):
00059 return
00060
00061
00062 def tearDown(self):
00063 OpenRTM_aist.Manager.instance().shutdownManager()
00064 return
00065
00066
00067 def test_addSdoServiceConsumerFactory(self):
00068 sdoadmin = SdoServiceAdmin(MockRTC())
00069 self.assertEqual(sdoadmin.addSdoServiceConsumerFactory(),False)
00070 return
00071
00072
00073 def test_removeSdoServiceConsumerFactory(self):
00074 sdoadmin = SdoServiceAdmin(MockRTC())
00075 self.assertEqual(sdoadmin.removeSdoServiceConsumerFactory(),False)
00076 return
00077
00078
00079 def test_addSdoServiceConsumer(self):
00080 any_val = any.to_any("3.14159")
00081 nv = SDOPackage.NameValue("PROPERTIES NAME 0", any_val)
00082 sprof = SDOPackage.ServiceProfile("ID 0","INTERFACE_TYPE",[nv],SDOServiceMock())
00083 sdoadmin = SdoServiceAdmin(MockRTC())
00084 self.assertEqual(sdoadmin.addSdoServiceConsumer(sprof),False)
00085 return
00086
00087
00088 def test_removeSdoServiceConsumer(self):
00089 any_val = any.to_any("3.14159")
00090 nv = SDOPackage.NameValue("PROPERTIES NAME 0", any_val)
00091 sprof = SDOPackage.ServiceProfile("test id","INTERFACE_TYPE",[nv],SDOServiceMock())
00092 sdoadmin = SdoServiceAdmin(MockRTC())
00093 self.assertEqual(sdoadmin.addSdoServiceConsumer(sprof),False)
00094 sdoadmin._consumers.append(MockSdoServiceConsumer())
00095 self.assertEqual(sdoadmin.removeSdoServiceConsumer("test id"),True)
00096 self.assertEqual(sdoadmin.addSdoServiceConsumer(sprof),False)
00097 sdoadmin._consumers.append(MockSdoServiceConsumer())
00098 self.assertEqual(sdoadmin.removeSdoServiceConsumer("test id2"),False)
00099 return
00100
00101
00102 def test_isAllowedConsumerType(self):
00103 sdoadmin = SdoServiceAdmin(MockRTC())
00104 any_val = any.to_any("3.14159")
00105 nv = SDOPackage.NameValue("PROPERTIES NAME 0", any_val)
00106 sprof = SDOPackage.ServiceProfile("test id","INTERFACE_TYPE",[nv],SDOServiceMock())
00107 self.assertEqual(sdoadmin.isAllowedConsumerType(sprof),True)
00108 sdoadmin._allConsumerAllowed = False
00109 self.assertEqual(sdoadmin.isAllowedConsumerType(sprof),False)
00110 sdoadmin._consumerTypes = ["type0","type1","type2","INTERFACE_TYPE"]
00111 self.assertEqual(sdoadmin.isAllowedConsumerType(sprof),True)
00112 sdoadmin._consumerTypes = ["type0","type1","type2"]
00113 self.assertEqual(sdoadmin.isAllowedConsumerType(sprof),False)
00114 return
00115
00116
00117 def test_isExistingConsumerType(self):
00118 factory = OpenRTM_aist.SdoServiceConsumerFactory.instance()
00119 factory.addFactory("test_factory",OpenRTM_aist.SdoServiceConsumerBase,OpenRTM_aist.Delete)
00120 sdoadmin = SdoServiceAdmin(MockRTC())
00121 any_val = any.to_any("3.14159")
00122 nv = SDOPackage.NameValue("PROPERTIES NAME 0", any_val)
00123 sprof = SDOPackage.ServiceProfile("test id","INTERFACE_TYPE",[nv],SDOServiceMock())
00124 self.assertEqual(sdoadmin.isExistingConsumerType(sprof),False)
00125 factory.addFactory("INTERFACE_TYPE",OpenRTM_aist.SdoServiceConsumerBase,OpenRTM_aist.Delete)
00126 self.assertEqual(sdoadmin.isExistingConsumerType(sprof),True)
00127 return
00128
00129
00130 def test_getUUID(self):
00131 sdoadmin = SdoServiceAdmin(MockRTC())
00132 self.assertNotEqual(sdoadmin.getUUID(),"")
00133 self.assertNotEqual(sdoadmin.getUUID(),None)
00134 return
00135
00136
00137
00138 if __name__ == '__main__':
00139 unittest.main()