#!/usr/bin/env python
"""
This module tests multiple message cyclic send tasks.
"""
import unittest
import time
import can
from .config import TEST_INTERFACE_SOCKETCAN
@unittest.skipUnless(TEST_INTERFACE_SOCKETCAN, "skip testing of socketcan")
class CyclicSocketCan(unittest.TestCase):
BITRATE = 500000
TIMEOUT = 0.1
INTERFACE_1 = "socketcan"
CHANNEL_1 = "vcan0"
INTERFACE_2 = "socketcan"
CHANNEL_2 = "vcan0"
PERIOD = 1.0
DELTA = 0.01
def _find_start_index(self, tx_messages, message):
"""
:param tx_messages:
The list of messages that were passed to the periodic backend
:param message:
The message whose data we wish to match and align to
:returns: start index in the tx_messages
"""
start_index = -1
for index, tx_message in enumerate(tx_messages):
if tx_message.data == message.data:
start_index = index
break
return start_index
def setUp(self):
self._send_bus = can.Bus(
interface=self.INTERFACE_1, channel=self.CHANNEL_1, bitrate=self.BITRATE
)
self._recv_bus = can.Bus(
interface=self.INTERFACE_2, channel=self.CHANNEL_2, bitrate=self.BITRATE
)
def tearDown(self):
self._send_bus.shutdown()
self._recv_bus.shutdown()
def test_cyclic_initializer_list(self):
messages = []
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x22, 0x22, 0x22, 0x22, 0x22, 0x22],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x33, 0x33, 0x33, 0x33, 0x33, 0x33],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x44, 0x44, 0x44, 0x44, 0x44, 0x44],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x55, 0x55, 0x55, 0x55, 0x55, 0x55],
is_extended_id=False,
)
)
task = self._send_bus.send_periodic(messages, self.PERIOD)
self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC)
results = []
for _ in range(len(messages) * 2):
result = self._recv_bus.recv(self.PERIOD * 2)
if result:
results.append(result)
task.stop()
# Find starting index for each
start_index = self._find_start_index(messages, results[0])
self.assertTrue(start_index != -1)
# Now go through the partitioned results and assert that they're equal
for rx_index, rx_message in enumerate(results):
tx_message = messages[start_index]
self.assertIsNotNone(rx_message)
self.assertEqual(tx_message.arbitration_id, rx_message.arbitration_id)
self.assertEqual(tx_message.dlc, rx_message.dlc)
self.assertEqual(tx_message.data, rx_message.data)
self.assertEqual(tx_message.is_extended_id, rx_message.is_extended_id)
self.assertEqual(tx_message.is_remote_frame, rx_message.is_remote_frame)
self.assertEqual(tx_message.is_error_frame, rx_message.is_error_frame)
self.assertEqual(tx_message.is_fd, rx_message.is_fd)
start_index = (start_index + 1) % len(messages)
def test_cyclic_initializer_tuple(self):
messages = []
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x22, 0x22, 0x22, 0x22, 0x22, 0x22],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x33, 0x33, 0x33, 0x33, 0x33, 0x33],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x44, 0x44, 0x44, 0x44, 0x44, 0x44],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x55, 0x55, 0x55, 0x55, 0x55, 0x55],
is_extended_id=False,
)
)
messages = tuple(messages)
self.assertIsInstance(messages, tuple)
task = self._send_bus.send_periodic(messages, self.PERIOD)
self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC)
results = []
for _ in range(len(messages) * 2):
result = self._recv_bus.recv(self.PERIOD * 2)
if result:
results.append(result)
task.stop()
# Find starting index for each
start_index = self._find_start_index(messages, results[0])
self.assertTrue(start_index != -1)
# Now go through the partitioned results and assert that they're equal
for rx_index, rx_message in enumerate(results):
tx_message = messages[start_index]
self.assertIsNotNone(rx_message)
self.assertEqual(tx_message.arbitration_id, rx_message.arbitration_id)
self.assertEqual(tx_message.dlc, rx_message.dlc)
self.assertEqual(tx_message.data, rx_message.data)
self.assertEqual(tx_message.is_extended_id, rx_message.is_extended_id)
self.assertEqual(tx_message.is_remote_frame, rx_message.is_remote_frame)
self.assertEqual(tx_message.is_error_frame, rx_message.is_error_frame)
self.assertEqual(tx_message.is_fd, rx_message.is_fd)
start_index = (start_index + 1) % len(messages)
def test_cyclic_initializer_message(self):
message = can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
task = self._send_bus.send_periodic(message, self.PERIOD)
self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC)
# Take advantage of kernel's queueing mechanisms
time.sleep(4 * self.PERIOD)
task.stop()
for _ in range(4):
tx_message = message
rx_message = self._recv_bus.recv(self.TIMEOUT)
self.assertIsNotNone(rx_message)
self.assertEqual(tx_message.arbitration_id, rx_message.arbitration_id)
self.assertEqual(tx_message.dlc, rx_message.dlc)
self.assertEqual(tx_message.data, rx_message.data)
self.assertEqual(tx_message.is_extended_id, rx_message.is_extended_id)
self.assertEqual(tx_message.is_remote_frame, rx_message.is_remote_frame)
self.assertEqual(tx_message.is_error_frame, rx_message.is_error_frame)
self.assertEqual(tx_message.is_fd, rx_message.is_fd)
def test_cyclic_initializer_invalid_none(self):
with self.assertRaises(ValueError):
task = self._send_bus.send_periodic(None, self.PERIOD)
def test_cyclic_initializer_invalid_empty_list(self):
with self.assertRaises(ValueError):
task = self._send_bus.send_periodic([], self.PERIOD)
def test_cyclic_initializer_different_arbitration_ids(self):
messages = []
messages.append(
can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
)
messages.append(
can.Message(
arbitration_id=0x3E1,
data=[0xEE, 0xEE, 0xEE, 0xEE, 0xEE, 0xEE],
is_extended_id=False,
)
)
with self.assertRaises(ValueError):
task = self._send_bus.send_periodic(messages, self.PERIOD)
def test_start_already_started_task(self):
messages_a = can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
task_a = self._send_bus.send_periodic(messages_a, self.PERIOD)
time.sleep(0.1)
# Task restarting is permitted as of #1440
task_a.start()
task_a.stop()
def test_create_same_id(self):
messages_a = can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
messages_b = can.Message(
arbitration_id=0x401,
data=[0x22, 0x22, 0x22, 0x22, 0x22, 0x22],
is_extended_id=False,
)
task_a = self._send_bus.send_periodic(messages_a, self.PERIOD)
self.assertIsInstance(task_a, can.broadcastmanager.CyclicSendTaskABC)
task_b = self._send_bus.send_periodic(messages_b, self.PERIOD)
self.assertIsInstance(task_b, can.broadcastmanager.CyclicSendTaskABC)
time.sleep(self.PERIOD * 4)
task_a.stop()
task_b.stop()
msgs = []
for _ in range(4):
msg = self._recv_bus.recv(self.PERIOD * 2)
self.assertIsNotNone(msg)
msgs.append(msg)
self.assertTrue(len(msgs) >= 4)
# Both messages should be recevied on the bus,
# even with the same arbitration id
msg_a_data_present = msg_b_data_present = False
for rx_message in msgs:
self.assertTrue(
rx_message.arbitration_id
== messages_a.arbitration_id
== messages_b.arbitration_id
)
if rx_message.data == messages_a.data:
msg_a_data_present = True
if rx_message.data == messages_b.data:
msg_b_data_present = True
self.assertTrue(msg_a_data_present)
self.assertTrue(msg_b_data_present)
def test_modify_data_list(self):
messages_odd = []
messages_odd.append(
can.Message(
arbitration_id=0x401,
data=[0x11, 0x11, 0x11, 0x11, 0x11, 0x11],
is_extended_id=False,
)
)
messages_odd.append(
can.Message(
arbitration_id=0x401,
data=[0x33, 0x33, 0x33, 0x33, 0x33, 0x33],
is_extended_id=False,
)
)
messages_odd.append(
can.Message(
arbitration_id=0x401,
data=[0x55, 0x55, 0x55, 0x55, 0x55, 0x55],
is_extended_id=False,
)
)
messages_even = []
messages_even.append(
can.Message(
arbitration_id=0x401,
data=[0x22, 0x22, 0x22, 0x22, 0x22, 0x22],
is_extended_id=False,
)
)
messages_even.append(
can.Message(
arbitration_id=0x401,
data=[0x44, 0x44, 0x44, 0x44, 0x44, 0x44],
is_extended_id=False,
)
)
messages_even.append(
can.Message(
arbitration_id=0x401,
data=[0x66, 0x66, 0x66, 0x66, 0x66, 0x66],
is_extended_id=False,
)
)
task = self._send_bus.send_periodic(messages_odd, self.PERIOD)
self.assertIsInstance(task, can.broadcastmanager.ModifiableCyclicTaskABC)
results_odd = []
results_even = []
for _ in range(len(messages_odd) * 2):
result = self._recv_bus.recv(self.PERIOD * 2)
if result:
results_odd.append(result)
task.modify_data(messages_even)
for _ in range(len(messages_even) * 2):
result = self._recv_bus.recv(self.PERIOD * 2)
if result:
results_even.append(result)
task.stop()
# Make sure we received some messages
self.assertTrue(len(results_even) != 0)
self.assertTrue(len(results_odd) != 0)
# Find starting index for each
start_index_even = self._find_start_index(messages_even, results_even[0])
self.assertTrue(start_index_even != -1)
start_index_odd = self._find_start_index(messages_odd, results_odd[0])
self.assertTrue(start_index_odd != -1)
# Now go through the partitioned results and assert that they're equal
for rx_index, rx_message in enumerate(results_even):
tx_message = messages_even[start_index_even]
self.assertEqual(tx_message.arbitration_id, rx_message.arbitration_id)
self.assertEqual(tx_message.dlc, rx_message.dlc)
self.assertEqual(tx_message.data, rx_message.data)
self.assertEqual(tx_message.is_extended_id, rx_message.is_extended_id)
self.assertEqual(tx_message.is_remote_frame, rx_message.is_remote_frame)
self.assertEqual(tx_message.is_error_frame, rx_message.is_error_frame)
self.assertEqual(tx_message.is_fd, rx_message.is_fd)
start_index_even = (start_index_even + 1) % len(messages_even)
if rx_index != 0:
prev_rx_message = results_even[rx_index - 1]
# Assert timestamps are within the expected period
self.assertTrue(
abs(
(rx_message.timestamp - prev_rx_message.timestamp) - self.PERIOD
)