test_PortAdmin.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- Python -*-

#
# \file test_PortAdmin.py
# \brief test for RTC's Port administration class
# \date $Date: 2007/09/03 $
# \author Shinji Kurihara
#
# Copyright (C) 2006
#     Task-intelligence Research Group,
#     Intelligent Systems Research Institute,
#     National Institute of
#         Advanced Industrial Science and Technology (AIST), Japan
#     All rights reserved.
#

import sys
sys.path.insert(1,"../")

import unittest
import OpenRTM_aist
from PortAdmin import *

import RTC, RTC__POA
import OpenRTM

from omniORB import CORBA

class PortBase(OpenRTM_aist.PortBase):
  def __init__(self):
    OpenRTM_aist.PortBase.__init__(self)

  def publishInterfaces(self, prof):
    return RTC.RTC_OK

  def subscribeInterfaces(self, prof):
    return RTC.RTC_OK

  def unsubscribeInterfaces(self, prof):
    return RTC.RTC_OK

  def activateInterfaces(self):
    print "activateInterfaces"


  def deactivateInterfaces(self):
    print "deactivateInterfaces"

    
class TestPortAdmin(unittest.TestCase):
  def setUp(self):
    self._orb = CORBA.ORB_init(sys.argv)
    self._poa = self._orb.resolve_initial_references("RootPOA")
    self._poa._get_the_POAManager().activate()

    self._pa = PortAdmin(self._orb, self._poa)

    self._pb1 = PortBase()
    self._pb2 = PortBase()

    self._pb1.setName("port0")
    self._pb2.setName("port1")

    self._pa.registerPort(self._pb1)
    self._pa.registerPort(self._pb2)
    return

  def tearDown(self):
    OpenRTM_aist.Manager.instance().shutdownManager()


  def test_getPortServiceList(self):
    plist = self._pa.getPortServiceList()

    prof0 = plist[0].get_port_profile()
    self.assertEqual(prof0.name, "port0")

    prof1 = plist[1].get_port_profile()
    self.assertEqual(prof1.name, "port1")
    return
    

  def test_getPortProfileList(self):
    pprof = self._pa.getPortProfileList()
    return


  def CCCtest_getPortRef(self):
    
    getP = self._pa.getPortRef("")
    self.assertEqual(CORBA.is_nil(getP), True)

    getP = self._pa.getPortRef("port1")
    self.assertEqual(CORBA.is_nil(getP), False)
    self.assertEqual(getP.get_port_profile().name, "port1")

    getP = self._pa.getPortRef("port0")
    self.assertEqual(CORBA.is_nil(getP), False)
    self.assertEqual(getP.get_port_profile().name, "port0")
    return
    

  def CCCtest_getPort(self):
    pb = self._pa.getPort("port0")
    prof = pb.get_port_profile()
    self.assertEqual(prof.name, "port0")

    pb = self._pa.getPort("port1")
    prof = pb.get_port_profile()
    self.assertEqual(prof.name, "port1")
    return


  def CCCtest_deletePort(self):
    self._pa.deletePort(self._pb1)
    plist = self._pa.getPortServiceList()
    self.assertEqual(len(plist), 1)

    prof = plist[0].get_port_profile()
    self.assertEqual(prof.name, "port1")
    return


  def CCCtest_deletePortByName(self):
    plist = self._pa.getPortServiceList()
    self.assertEqual(len(plist), 2)

    self._pa.deletePortByName("port1")

    plist = self._pa.getPortServiceList()
    self.assertEqual(len(plist), 1)
    return
    

  def CCCtest_activatePorts(self):
    self._pa.activatePorts()
    self._pa.deactivatePorts()
    return


  def CCCtest_finalizePorts(self):
    plist = self._pa.getPortServiceList()
    self.assertEqual(len(plist), 2)

    self._pa.finalizePorts()
    
    plist = self._pa.getPortServiceList()
    self.assertEqual(len(plist), 0)
    return
    



if __name__ == '__main__':
        unittest.main()