iframe-proxy | Sunbelt Computer Software

217 lines (174 loc) · 7.28 KB
#!/usr/bin/env python
# coding: utf-8
This module tests two virtual busses attached to each other.
from __future__ import absolute_import, print_function
import unittest
from time import sleep
from multiprocessing.dummy import Pool as ThreadPool 
import pytest
from .config import *
class Back2BackTestCase(unittest.TestCase):
    Use two interfaces connected to the same CAN bus and test them against
    each other.
    BITRATE = 500000
    TIMEOUT = 0.1
    INTERFACE_1 = 'virtual'
    CHANNEL_1 = 'virtual_channel_0'
    INTERFACE_2 = 'virtual'
    CHANNEL_2 = 'virtual_channel_0'
    def setUp(self):
        self.bus1 = can.Bus(channel=self.CHANNEL_1,
                            bustype=self.INTERFACE_1,
                            bitrate=self.BITRATE,
                            fd=TEST_CAN_FD,
                            single_handle=True)
        self.bus2 = can.Bus(channel=self.CHANNEL_2,
                            bustype=self.INTERFACE_2,
                            bitrate=self.BITRATE,
                            fd=TEST_CAN_FD,
                            single_handle=True)
    def tearDown(self):
        self.bus1.shutdown()
        self.bus2.shutdown()
    def _check_received_message(self, recv_msg, sent_msg):
        self.assertIsNotNone(recv_msg,
                             "No message was received on %s" % self.INTERFACE_2)
        self.assertEqual(recv_msg.arbitration_id, sent_msg.arbitration_id)
        self.assertEqual(recv_msg.id_type, sent_msg.id_type)
        self.assertEqual(recv_msg.is_remote_frame, sent_msg.is_remote_frame)
        self.assertEqual(recv_msg.is_error_frame, sent_msg.is_error_frame)
        self.assertEqual(recv_msg.is_fd, sent_msg.is_fd)
        self.assertEqual(recv_msg.bitrate_switch, sent_msg.bitrate_switch)
        self.assertEqual(recv_msg.dlc, sent_msg.dlc)
        if not sent_msg.is_remote_frame:
            self.assertSequenceEqual(recv_msg.data, sent_msg.data)
    def _send_and_receive(self, msg):
        # Send with bus 1, receive with bus 2
        self.bus1.send(msg)
        recv_msg = self.bus2.recv(self.TIMEOUT)
        self._check_received_message(recv_msg, msg)
        # Some buses may receive their own messages. Remove it from the queue
        self.bus1.recv(0)
        # 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(self.TIMEOUT)
        self._check_received_message(recv_msg, msg)
    def test_no_message(self):
        self.assertIsNone(self.bus1.recv(0.1))
    @unittest.skipIf(IS_CI, "the timing sensitive behaviour cannot be reproduced reliably on a CI server")
    def test_timestamp(self):
        self.bus2.send(can.Message())
        recv_msg1 = self.bus1.recv(self.TIMEOUT)
        sleep(2.0)
        self.bus2.send(can.Message())
        recv_msg2 = self.bus1.recv(self.TIMEOUT)
        delta_time = recv_msg2.timestamp - recv_msg1.timestamp
        self.assertTrue(1.75 <= delta_time <= 2.25,
                        'Time difference should have been 2s +/- 250ms.' 
                        'But measured {}'.format(delta_time))
    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,
        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)
    @unittest.skipUnless(TEST_CAN_FD, "Don't test CAN-FD")
    def test_fd_message(self):
        msg = can.Message(is_fd=True,
                          extended_id=True,
                          arbitration_id=0x56789,
                          data=[0xff] * 64)
        self._send_and_receive(msg)
    @unittest.skipUnless(TEST_CAN_FD, "Don't test CAN-FD")
    def test_fd_message_with_brs(self):
        msg = can.Message(is_fd=True,
                          bitrate_switch=True,
                          extended_id=True,
                          arbitration_id=0x98765,
                          data=[0xff] * 48)
        self._send_and_receive(msg)
@unittest.skipUnless(TEST_INTERFACE_SOCKETCAN, "skip testing of socketcan")
class BasicTestSocketCan(Back2BackTestCase):
    INTERFACE_1 = 'socketcan'
    CHANNEL_1 = 'vcan0'
    INTERFACE_2 = 'socketcan'
    CHANNEL_2 = 'vcan0'
@unittest.skipUnless(TEST_INTERFACE_SOCKETCAN, "skip testing of socketcan")
class SocketCanBroadcastChannel(unittest.TestCase):
    def setUp(self):
        self.broadcast_bus = can.Bus(channel='', bustype='socketcan')
        self.regular_bus = can.Bus(channel='vcan0', bustype='socketcan')
    def tearDown(self):
        self.broadcast_bus.shutdown()
        self.regular_bus.shutdown()
    def test_broadcast_channel(self):
        self.broadcast_bus.send(can.Message(channel='vcan0'))
        recv_msg = self.regular_bus.recv(1)
        self.assertIsNotNone(recv_msg)
        self.assertEqual(recv_msg.channel, 'vcan0')
        self.regular_bus.send(can.Message())
        recv_msg = self.broadcast_bus.recv(1)
        self.assertIsNotNone(recv_msg)
        self.assertEqual(recv_msg.channel, 'vcan0')
class TestThreadSafeBus(Back2BackTestCase):
    """Does some testing that is better than nothing.
    def setUp(self):
        self.bus1 = can.ThreadSafeBus(channel=self.CHANNEL_1,
                                      bustype=self.INTERFACE_1,
                                      bitrate=self.BITRATE,
                                      fd=TEST_CAN_FD,
                                      single_handle=True)
        self.bus2 = can.ThreadSafeBus(channel=self.CHANNEL_2,
                                      bustype=self.INTERFACE_2,
                                      bitrate=self.BITRATE,
                                      fd=TEST_CAN_FD,
                                      single_handle=True)
    @pytest.mark.timeout(5.0)
    def test_concurrent_writes(self):
        sender_pool = ThreadPool(100)
        receiver_pool = ThreadPool(100)
        message = can.Message(
            arbitration_id=0x123,
            extended_id=True,
            timestamp=121334.365,
            data=[254, 255, 1, 2]
        workload = 1000 * [message]
        def sender(msg):
            self.bus1.send(msg)
        def receiver(_):
            result = self.bus2.recv(timeout=2.0)
            self.assertIsNotNone(result)
            self.assertEqual(result, message)
        sender_pool.map_async(sender, workload)
        receiver_pool.map_async(receiver, len(workload) * [None])
        sender_pool.close()
        sender_pool.join()
        receiver_pool.close()
        receiver_pool.join()
if __name__ == '__main__':
    unittest.main()
Sunbelt Computer Software

PL/B Language Development and Support

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Sunbelt Computer Software

PL/B Language Development and Support

FilesExpand file tree

back2back_test.py

Latest commit

History

back2back_test.py

File metadata and controls
