Merge pull request #25 from jamesleesaunders/fake-serial

Improved Fake Serial device used by unit tests

Author: hansmosh

xbee/tests/Fake.py

@@ -4,53 +4,123 @@
 
 By Paul Malmsten, 2010
 pmalmsten@gmail.com
+Updated by James Saunders, 2016
+Inspired by code written by D. Thiebaut http://cs.smith.edu/dftwiki/index.php/PySerial_Simulator
 
 Provides fake device objects for other unit tests.
 """
 import sys
 
-class FakeDevice(object):
-    """
-    Represents a fake serial port for testing purposes
-    """
-    def __init__(self):
-        self.data = b''
-    
-    def write(self, data):
+class Serial(object):
+    def __init__( self, port='/dev/null', baudrate = 19200, timeout=1,
+                  bytesize = 8, parity = 'N', stopbits = 1, xonxoff=0,
+                  rtscts = 0 ):
         """
-        Writes data to the fake port for later evaluation
+        Init constructor, setup standard serial variables with default values.
         """
-        self.data = data
+        self.name     = port
+        self.port     = port
+        self.timeout  = timeout
+        self.parity   = parity
+        self.baudrate = baudrate
+        self.bytesize = bytesize
+        self.stopbits = stopbits
+        self.xonxoff  = xonxoff
+        self.rtscts   = rtscts
+        self._is_open = True        
 
-class FakeReadDevice(object):
-    """
-    Represents a fake serial port which can be read from in a similar
-    fashion to the real thing
-    """
-    
-    def __init__(self, data, silent_on_empty=False):
-        self.data = data
-        self.read_index = 0
-        self.silent_on_empty = silent_on_empty
-        
-    def read(self, length=1):
+        self._data_written = ""
+        self._read_data    = ""
+
+    def isOpen( self ):
         """
-        Read the indicated number of bytes from the port
+        Returns True if the serial port is open, otherwise False.
         """
-        # If too many bytes would be read, raise exception
-        if self.read_index + length > len(self.data):
-            if self.silent_on_empty:
-                sys.exit(0)
-            else:
-                raise ValueError("Not enough bytes exist!")
-        
-        read_data = self.data[self.read_index:self.read_index + length]
-        self.read_index += length
-        
-        return read_data
+        return self._isOpen
+
+    def open( self ):
+        """
+        Open the serial port.
+        """
+        self._is_open = True
+
+    def close( self ):
+        """
+        Close the serial port.
+        """
+        self._is_open = False
+
+    def write( self, data ):
+        """
+        Write a string of characters to the serial port.
+        """
+        self._data_written = data
+
+    def read( self, len=1 ):
+        """
+        Read the indicated number of bytes from the port.
+        """
+        data = self._read_data[0:len]
+        self._read_data = self._read_data[len:]
+        return data
+
+    def readline( self ):
+        """
+        Read characters from the port until a '\n' (newline) is found.
+        """
+        returnIndex = self._read_data.index( "\n" )
+        if returnIndex != -1:
+            data = self._read_data[0:returnIndex+1]
+            self._read_data = self._read_data[returnIndex+1:]
+            return data
+        else:
+            return ""
+
+    def inWaiting( self ):
+        """
+        Returns the number of bytes available to be read.
+        """
+        return len(self._read_data)
+
+    def getSettingsDict( self ):
+        """"
+        Get a dictionary with port settings.
+        """
+        settings = {
+            'timeout'  : self.timeout, 
+            'parity'   : self.parity, 
+            'baudrate' : self.baudrate, 
+            'bytesize' : self.bytesize,
+            'stopbits' : self.stopbits, 
+            'xonxoff'  : self.xonxoff, 
+            'rtscts'   : self.rtscts 
+        }
+        return settings
+
+    def set_read_data( self, data ):
+        """
+        Set fake data to be be returned by the read() and readline() functions.
+        """
+        self._read_data = data
+
+    def get_data_written( self ):
+        """
+        Return record of data sent via the write command.
+        """
+        return(self._data_written)
+
+    def set_silent_on_empty( self, flag ):
+        """
+        Set silent on error flag. If True do not error.
+        """
+        self._silent_on_empty = flag
 
-    def inWaiting(self):
+    def __str__( self ):
         """
-        Returns the number of bytes available to be read
+        Returns a string representation of the serial class.
         """
-        return len(self.data) - self.read_index
+        return  "Serial<id=0xa81c10, open=%s>( port='%s', baudrate=%d," \
+               % ( str(self._is_open), self.port, self.baudrate ) \
+               + " bytesize=%d, parity='%s', stopbits=%d, xonxoff=%d, rtscts=%d)"\
+               % ( self.bytesize, self.parity, self.stopbits, self.xonxoff,
+                   self.rtscts )

xbee/tests/test_base.py

@@ -9,7 +9,7 @@
 """
 import unittest
 from xbee.base import XBeeBase
-from xbee.tests.Fake import FakeDevice, FakeReadDevice
+from xbee.tests.Fake import Serial
 
 class TestWriteToDevice(unittest.TestCase):
     """
@@ -22,42 +22,45 @@ def test_write(self):
         _write method should write the expected data to the serial
         device
         """
-        device = FakeDevice()
+        device = Serial()
 
         xbee = XBeeBase(device)
         xbee._write(b'\x00')
 
         # Check resuting state of fake device
+        result_frame   = device.get_data_written()
         expected_frame = b'\x7E\x00\x01\x00\xFF'
-        self.assertEqual(device.data, expected_frame)
+        self.assertEqual(result_frame, expected_frame)
 
     def test_write_again(self):
         """
         _write method should write the expected data to the serial
         device
         """
-        device = FakeDevice()
+        device = Serial()
 
         xbee = XBeeBase(device)
         xbee._write(b'\x00\x01\x02')
 
         # Check resuting state of fake device
         expected_frame = b'\x7E\x00\x03\x00\x01\x02\xFC'
-        self.assertEqual(device.data, expected_frame)
+        result_frame   = device.get_data_written()
+        self.assertEqual(result_frame, expected_frame)
 
     def test_write_escaped(self):
         """
         _write method should write the expected data to the serial
         device
         """
-        device = FakeDevice()
+        device = Serial()
 
         xbee = XBeeBase(device,escaped=True)
         xbee._write(b'\x7E\x01\x7D\x11\x13')
 
         # Check resuting state of fake device
         expected_frame = b'\x7E\x00\x05\x7D\x5E\x01\x7D\x5D\x7D\x31\x7D\x33\xDF'
-        self.assertEqual(device.data, expected_frame)
+        result_frame   = device.get_data_written()
+        self.assertEqual(result_frame, expected_frame)
 
 class TestReadFromDevice(unittest.TestCase):
     """
@@ -68,7 +71,8 @@ def test_read(self):
         """
         _wait_for_frame should properly read a frame of data
         """
-        device = FakeReadDevice(b'\x7E\x00\x01\x00\xFF')
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x01\x00\xFF')
         xbee = XBeeBase(device)
 
         frame = xbee._wait_for_frame()
@@ -78,8 +82,8 @@ def test_read_invalid_followed_by_valid(self):
         """
         _wait_for_frame should skip invalid data
         """
-        device = FakeReadDevice(
-            b'\x7E\x00\x01\x00\xFA' + b'\x7E\x00\x01\x05\xFA')
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x01\x00\xFA' + b'\x7E\x00\x01\x05\xFA')
         xbee = XBeeBase(device)
 
         frame = xbee._wait_for_frame()
@@ -90,7 +94,8 @@ def test_read_escaped(self):
         _wait_for_frame should properly read a frame of data
         Verify that API mode 2 escaped bytes are read correctly
         """
-        device = FakeReadDevice(b'\x7E\x00\x04\x7D\x5E\x7D\x5D\x7D\x31\x7D\x33\xE0')
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x04\x7D\x5E\x7D\x5D\x7D\x31\x7D\x33\xE0')
 
         xbee = XBeeBase(device,escaped=True)
 
@@ -150,7 +155,7 @@ class TestAsyncCallback(unittest.TestCase):
 
     def setUp(self):
         self.xbee = None
-        self.serial = FakeReadDevice([], silent_on_empty=True)
+        self.serial = Serial()
         self.callback = lambda data: None
         self.error_callback = lambda data: None
 

xbee/tests/test_fake.py

@@ -8,22 +8,23 @@
 Tests fake device objects for proper functionality.
 """
 import unittest
-from xbee.tests.Fake import FakeReadDevice
+from xbee.tests.Fake import Serial
 
-class TestFakeReadDevice(unittest.TestCase):
+class TestFakeSerialRead(unittest.TestCase):
     """
-    FakeReadDevice class should work as intended to emluate a serial 
-    port
+    Fake Serial class should work as intended to emluate reading from a serial port.
     """
+
     def setUp(self):
         """
-        Create a fake read device for each test
+        Create a fake read device for each test.
         """
-        self.device = FakeReadDevice("test")
-    
+        self.device = Serial()
+        self.device.set_read_data("test")
+
     def test_read_single_byte(self):
         """
-        reading one byte at a time should work as expected
+        Reading one byte at a time should work as expected.
         """
         self.assertEqual(self.device.read(), 't')
         self.assertEqual(self.device.read(), 'e')
@@ -32,13 +33,14 @@ def test_read_single_byte(self):
 
     def test_read_multiple_bytes(self):
         """
-        reading multiple bytes at a time should work as expected
+        Reading multiple bytes at a time should work as expected.
         """
         self.assertEqual(self.device.read(3), 'tes')
         self.assertEqual(self.device.read(), 't')
 
-    def test_read_too_many(self):
+    def test_write(self):
         """
-        attempting to read too many bytes should raise an exception
+        Test serial write function.
         """
-        self.assertRaises(ValueError, self.device.read, 5)
+        self.device.write("Hello World")
+        self.assertEqual(self.device.get_data_written(), "Hello World")