Add back to back test with two interfaces

christiansandberg · christiansandberg · commit 8eb6a4e2b313 · 2017-07-07T21:58:59.000+02:00
Can be used to compare two interfaces that are connected to the same CAN bus
diff --git a/test/back2back_test.py b/test/back2back_test.py
@@ -0,0 +1,77 @@
+import unittest
+
+import can
+
+
+BITRATE = 500000
+TIMEOUT = 0.1
+
+INTERFACE_1 = 'virtual'
+CHANNEL_1 = 0
+INTERFACE_2 = 'virtual'
+CHANNEL_2 = 0
+
+
+class Back2BackTestCase(unittest.TestCase):
+    """
+    Use two interfaces connected to the same CAN bus and test them against
+    each other.
+    """
+
+    def setUp(self):
+        self.bus1 = can.interface.Bus(channel=CHANNEL_1,
+                                      bustype=INTERFACE_1,
+                                      bitrate=BITRATE)
+        self.bus2 = can.interface.Bus(channel=CHANNEL_2,
+                                      bustype=INTERFACE_2,
+                                      bitrate=BITRATE)
+
+    def tearDown(self):
+        self.bus1.shutdown()
+        self.bus2.shutdown()
+
+    def _send_and_receive(self, msg):
+        # Send with bus 1, receive with bus 2
+        self.bus1.send(msg)
+        recv_msg = self.bus2.recv(TIMEOUT)
+        self.assertIsNotNone(recv_msg,
+                             "No message was received on %s" % INTERFACE_2)
+        self.assertEqual(recv_msg, msg)
+
+        # Send with bus 2, receive with bus 1
+        # Add 1 to arbitration ID to make it a different message
+        msg.arbitration_id += 1
+        self.bus2.send(msg)
+        recv_msg = self.bus1.recv(TIMEOUT)
+        self.assertIsNotNone(recv_msg,
+                             "No message was received on %s" % INTERFACE_1)
+        self.assertEqual(recv_msg, msg)
+
+    def test_standard_message(self):
+        msg = can.Message(extended_id=False,
+                          arbitration_id=0x100,
+                          data=[1, 2, 3, 4, 5, 6, 7, 8])
+        self._send_and_receive(msg)
+
+    def test_extended_message(self):
+        msg = can.Message(extended_id=True,
+                          arbitration_id=0x123456,
+                          data=[10, 11, 12, 13, 14, 15, 16, 17])
+        self._send_and_receive(msg)
+
+    def test_remote_message(self):
+        msg = can.Message(extended_id=False,
+                          arbitration_id=0x200,
+                          is_remote_frame=True,
+                          dlc=4)
+        self._send_and_receive(msg)
+
+    def test_dlc_less_than_eight(self):
+        msg = can.Message(extended_id=False,
+                          arbitration_id=0x300,
+                          data=[4, 5, 6])
+        self._send_and_receive(msg)
+
+
+if __name__ == '__main__':
+    unittest.main()
