test_OutPortCorbaCdrProvider.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- Python -*-
 
#
#  \file  test_OutPortCorbaCdrProvider.py
#  \brief test for OutPortCorbaCdrProvider class
#  \date  $Date: 2007/09/26 $
#  \author Shinji Kurihara
# 
#  Copyright (C) 2006
#      Noriaki Ando
#      Task-intelligence Research Group,
#      Intelligent Systems Research Institute,
#      National Institute of
#          Advanced Industrial Science and Technology (AIST), Japan
#      All rights reserved.
 
from omniORB import any
from omniORB import CORBA
 
import OpenRTM_aist
import RTC, RTC__POA 
import SDOPackage, SDOPackage__POA
 
import sys
sys.path.insert(1,"../")
 
import unittest
 
from OutPortCorbaCdrProvider import *
 
class DummyBuffer:
  def __init__(self):
    self._cdr = None
    self._empty = True
 
  def empty(self):
    return self._empty
 
  def write(self,d):
    self._cdr = d
    self._empty = False
    return 0
 
  def read(self,cdr):
    cdr[0] = self._cdr
    self._empty = True
    return 0
 
class TestOutPortCorbaCdrProvider(unittest.TestCase):
 
  def setUp(self):
    OpenRTM_aist.Manager.instance()
    OpenRTM_aist.OutPortCorbaCdrProviderInit()
    self._opp = OpenRTM_aist.OutPortCorbaCdrProvider()
    return
    
  def test_setBuffer(self):
    self._opp.setBuffer(DummyBuffer())
    return
 
  def test_get(self):
    ret,data=self._opp.get()
    self.assertEqual(ret,OpenRTM.UNKNOWN_ERROR)
 
    prop = OpenRTM_aist.Properties()
    cinfo = OpenRTM_aist.ConnectorInfo("",
                                       "",
                                       [],
                                       prop)
    self._opp.setListener(cinfo,OpenRTM_aist.ConnectorListeners())
    buff = DummyBuffer()
    self._opp.setBuffer(buff)
    ret,data=self._opp.get()
    self.assertEqual(ret,OpenRTM.BUFFER_EMPTY)
 
    buff.write(123)
    ret,data=self._opp.get()
    self.assertEqual(data,123)
    return
 
 
 
if __name__ == '__main__':
        unittest.main()