Making test_base and test_ieee work with new Fake Serial class

* Replaced the previous FakeDevice and FakeReadDevice with the new fake Serial class.
* Use of get_data_written().
* Modify BadReadDevice accordingly.
* Adding default second argument 'length' to BadReadDevice().
* Removed test_read_too_many() test as no longer required.

Author: jamesleesaunders

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_ieee.py

@@ -8,7 +8,7 @@
 Tests the XBee (IEEE 802.15.4) implementation class for XBee API compliance
 """
 import unittest, sys, traceback
-from xbee.tests.Fake import FakeDevice, FakeReadDevice
+from xbee.tests.Fake import Serial
 from xbee.ieee import XBee
 from xbee.frame import APIFrame
 from xbee.python2to3 import byteToInt, intToByte, stringToBytes
@@ -443,15 +443,16 @@ def test_send_at_command(self):
         API AT command packet to the serial device.
         """
 
-        serial_port = FakeDevice()
+        serial_port = Serial()
         xbee = XBee(serial_port)
 
         # Send an AT command
         xbee.send('at', frame_id=stringToBytes('A'), command=stringToBytes('MY'))
 
         # Expect a full packet to be written to the device
         expected_data = b'\x7E\x00\x04\x08AMY\x10'
-        self.assertEqual(serial_port.data, expected_data)
+        result_data   = serial_port.get_data_written()
+        self.assertEqual(result_data, expected_data)
 
 
     def test_send_at_command_with_param(self):
@@ -460,7 +461,7 @@ def test_send_at_command_with_param(self):
         API AT command packet to the serial device.
         """
 
-        serial_port = FakeDevice()
+        serial_port = Serial()
         xbee = XBee(serial_port)
 
         # Send an AT command
@@ -472,8 +473,9 @@ def test_send_at_command_with_param(self):
         )
 
         # Expect a full packet to be written to the device
+        result_data   = serial_port.get_data_written()
         expected_data = b'\x7E\x00\x06\x08AMY\x00\x00\x10'
-        self.assertEqual(serial_port.data, expected_data)
+        self.assertEqual(result_data, expected_data)
 
 class TestSendShorthand(unittest.TestCase):
     """
@@ -485,7 +487,7 @@ def setUp(self):
         """
         Prepare a fake device to read from
         """
-        self.ser = FakeDevice()
+        self.ser = Serial()
         self.xbee = XBee(self.ser)
 
     def test_send_at_command(self):
@@ -496,30 +498,33 @@ def test_send_at_command(self):
         self.xbee.at(frame_id=stringToBytes('A'), command=stringToBytes('MY'))
 
         # Expect a full packet to be written to the device
+        result_data = self.ser.get_data_written()
         expected_data = b'\x7E\x00\x04\x08AMY\x10'
-        self.assertEqual(self.ser.data, expected_data)
+        self.assertEqual(result_data, expected_data)
 
     def test_send_at_command_with_param(self):
         """
         calling send should write a full API frame containing the
         API AT command packet to the serial device.
         """
-        
+       
         # Send an AT command
         self.xbee.at(frame_id=stringToBytes('A'), command=stringToBytes('MY'), parameter=b'\x00\x00')
 
         # Expect a full packet to be written to the device
+        result_data = self.ser.get_data_written()
         expected_data = b'\x7E\x00\x06\x08AMY\x00\x00\x10'
-        self.assertEqual(self.ser.data, expected_data)
+        self.assertEqual(result_data, expected_data)
 
     def test_send_tx_with_close_brace(self):
         """
         Calling tx where the given data string includes a close brace '}'
         must write correctly.
         """
         self.xbee.tx(dest_addr=b'\x01\x02',data=b'{test=1}')
+        result_data = self.ser.get_data_written()
         expected_data = b'\x7E\x00\x0D\x01\x00\x01\x02\x00{test=1}\xD5'
-        self.assertEqual(self.ser.data, expected_data)
+        self.assertEqual(result_data, expected_data)
 
     def test_shorthand_disabled(self):
         """
@@ -544,7 +549,8 @@ def test_read_at(self):
         """
         read and parse a parameterless AT command
         """
-        device = FakeReadDevice(b'\x7E\x00\x05\x88DMY\x01\x8c')
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x05\x88DMY\x01\x8c')
         xbee = XBee(device)
 
         info = xbee.wait_read_frame()
@@ -558,9 +564,8 @@ def test_read_at_params(self):
         """
         read and parse an AT command with a parameter
         """
-        device = FakeReadDevice(
-            b'\x7E\x00\x08\x88DMY\x01\x00\x00\x00\x8c'
-        )
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x08\x88DMY\x01\x00\x00\x00\x8c')
         xbee = XBee(device)
 
         info = xbee.wait_read_frame()
@@ -585,10 +590,8 @@ def test_is_response_parsed_as_io(self):
         sample = b'\x00\xAA\x00\xFF'
         data = header + sample
 
-        device = FakeReadDevice(
-            APIFrame(data = b'\x88DIS\x00' + data).output()
-        )
-        
+        device = Serial()
+        device.set_read_data(APIFrame(data = b'\x88DIS\x00' + data).output());
         xbee = XBee(device)
 
         info = xbee.wait_read_frame()
@@ -616,10 +619,9 @@ def test_is_remote_response_parsed_as_io(self):
         # ADC0 value of 255
         sample = b'\x00\xAA\x00\xFF'
         data = header + sample
-        
-        device = FakeReadDevice(
-            APIFrame(data = b'\x97D\x00\x13\xa2\x00@oG\xe4v\x1aIS\x00' + data).output()
-        )
+       
+        device = Serial()
+        device.set_read_data(APIFrame(data = b'\x97D\x00\x13\xa2\x00@oG\xe4v\x1aIS\x00' + data).output())
 
         xbee = XBee(device)
 
@@ -655,9 +657,8 @@ def test_read_io_data(self):
         # RX frame data
         rx_io_resp = b'\x83\x00\x01\x28\x00'
 
-        device = FakeReadDevice(
-            b'\x7E\x00\x0C'+ rx_io_resp + data + b'\xfd'
-        )
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x0C'+ rx_io_resp + data + b'\xfd')
         xbee = XBee(device)
 
         info = xbee.wait_read_frame()
@@ -681,16 +682,17 @@ def test_read_empty_string(self):
         an empty string. In this event, we must not crash.
         """
 
-        class BadReadDevice(FakeReadDevice):
+        class BadReadDevice(Serial):
             def __init__(self, bad_read_index, data):
                 self.read_id = 0
                 self.bad_read_index = bad_read_index
-                super(BadReadDevice, self).__init__(data)
+                super(BadReadDevice, self).__init__()
+                self.set_read_data(data)
 
-            def inWaiting(self):
+            def in_waiting(self):
                 return 1
 
-            def read(self):
+            def read(self, length=1):
                 if self.read_id == self.bad_read_index:
                     self.read_id += 1
                     return ''
@@ -711,9 +713,8 @@ def test_read_at_params_in_escaped_mode(self):
         """
         read and parse an AT command with a parameter in escaped API mode
         """
-        device = FakeReadDevice(
-            b'~\x00\t\x88DMY\x01}^}]}1}3m'
-        )
+        device = Serial()
+        device.set_read_data(b'~\x00\t\x88DMY\x01}^}]}1}3m')
         xbee = XBee(device, escaped=True)
 
         info = xbee.wait_read_frame()
@@ -728,7 +729,8 @@ def test_empty_frame_ignored(self):
         """
         If an empty frame is received from a device, it must be ignored.
         """
-        device = FakeReadDevice(b'\x7E\x00\x00\xFF\x7E\x00\x05\x88DMY\x01\x8c')
+        device = Serial()
+        device.set_read_data(b'\x7E\x00\x00\xFF\x7E\x00\x05\x88DMY\x01\x8c')
         xbee = XBee(device)
 
         #import pdb
@@ -744,7 +746,8 @@ def test_read_rx_with_close_brace(self):
         """
         An rx data frame including a close brace must be read properly.
         """
-        device = FakeReadDevice(APIFrame(b'\x81\x01\x02\x55\x00{test=1}').output())
+        device = Serial()
+        device.set_read_data(APIFrame(b'\x81\x01\x02\x55\x00{test=1}').output())
         xbee = XBee(device)
 
         info = xbee.wait_read_frame()
@@ -759,9 +762,8 @@ def test_read_rx_with_close_brace_escaped(self):
         """
         An escaped rx data frame including a close brace must be read properly.
         """
-        device = FakeReadDevice(
-            APIFrame(b'\x81\x01\x02\x55\x00{test=1}', escaped=True).output()
-        )
+        device = Serial()
+        device.set_read_data(APIFrame(b'\x81\x01\x02\x55\x00{test=1}', escaped=True).output())
         xbee = XBee(device, escaped=True)
 
         info = xbee.wait_read_frame()