test_InPort.py

Go to the documentation of this file.

#!/usr/bin/env python
# -*- Python -*-
 
 
#  \file test_InPort.py
#  \brief test for InPort template class
#  \date $Date: 2007/09/20 $
#  \author Shinji Kurihara
# 
#  Copyright (C) 2003-2005
#      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
 
from InPort import *
 
import RTC, RTC__POA
import OpenRTM_aist
 
ShareCount = 0
 
class OnRWTest:
  def __init__(self):
    pass
 
  def echo(self, value=None):
    print("OnRW Called")
 
 
class OnRWConvertTest:
  def __init__(self):
    pass
 
  def echo(self, value=None):
    print("OnRWConvert Called")
    return value
 
 
class BufferMock:
  def __init__(self):
    self._ShareCount = 0
    self._data = None
    return
 
  def write(self, data):
    global ShareCount
    ShareCount += 1
    self._ShareCount = ShareCount
    self._data = data
    return OpenRTM_aist.BufferStatus.BUFFER_OK
 
  def read(self, value):
    global ShareCount
    ShareCount -= 1
    if ShareCount < 0:
      ShareCount = 0
    self._ShareCount = ShareCount
    if len(value) > 0:
      value[0] = self._data
    else:
      value.append(self._data)
    return OpenRTM_aist.BufferStatus.BUFFER_OK
 
  def readable(self):
    return self._ShareCount
 
 
class ConnectorMock:
  def __init__(self, buffer):
    self._buffer = buffer
    return
 
  def write(self, data):
    self._buffer.write(data)
    return OpenRTM_aist.BufferStatus.BUFFER_OK
 
  def read(self, data):
    self._buffer.read(data)
    return OpenRTM_aist.DataPortStatus.PORT_OK
 
  def getBuffer(self):
    ret = self._buffer.readable()
    return self._buffer
 
 
class TestInPort(unittest.TestCase):
  def setUp(self):
    OpenRTM_aist.Manager.instance()
    self._buffer = BufferMock()
    self._ipn = InPort("in", RTC.TimedLong(RTC.Time(0,0), 0), self._buffer)
    self._connector = ConnectorMock(self._buffer)
    self._ipn._connectors = [self._connector]
 
  def tearDown(self):
    OpenRTM_aist.Manager.instance().shutdownManager()
    return
 
  def test_name(self):
    self.assertEqual(self._ipn.name(), "in")
 
  def test_read(self):
    self.assertEqual(self._ipn.isEmpty(), True)
    self.assertEqual(self._ipn.isNew(), False)
    self._connector.write(RTC.TimedLong(RTC.Time(0,0), 123))
    self.assertEqual(self._ipn.isEmpty(), False)
    self.assertEqual(self._ipn.isNew(), True)
    read_data = self._ipn.read()
    self.assertEqual(self._ipn.isEmpty(), True)
    self.assertEqual(self._ipn.isNew(), False)
    self.assertEqual(read_data.data, 123)
    self._ipn.update()
    return
 
  def test_OnRead(self):
    self._connector.write(RTC.TimedLong(RTC.Time(0,0), 456))
    self._ipn.setOnRead(OnRWTest().echo)
    self._ipn.setOnReadConvert(OnRWConvertTest().echo)
    read_data = self._ipn.read()
    self.assertEqual(read_data.data, 456)
    return
 
 
 
if __name__ == '__main__':
  unittest.main()