#!/usr/bin/env python """ This module tests cyclic send tasks. """ import gc import time import unittest from time import sleep from typing import List from unittest.mock import MagicMock import can from .config import * from .message_helper import ComparingMessagesTestCase class SimpleCyclicSendTaskTest(unittest.TestCase, ComparingMessagesTestCase): def __init__(self, *args, **kwargs): unittest.TestCase.__init__(self, *args, **kwargs) ComparingMessagesTestCase.__init__( self, allowed_timestamp_delta=0.016, preserves_channel=True ) @unittest.skipIf( IS_CI, "the timing sensitive behaviour cannot be reproduced reliably on a CI server", ) def test_cycle_time(self): msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7] ) with can.interface.Bus(interface="virtual") as bus1: with can.interface.Bus(interface="virtual") as bus2: # disabling the garbage collector makes the time readings more reliable gc.disable() task = bus1.send_periodic(msg, 0.01, 1) self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC) sleep(2) size = bus2.queue.qsize() # About 100 messages should have been transmitted self.assertTrue( 80 <= size <= 120, "100 +/- 20 messages should have been transmitted. But queue contained {}".format( size ), ) last_msg = bus2.recv() next_last_msg = bus2.recv() # we need to reenable the garbage collector again gc.enable() # Check consecutive messages are spaced properly in time and have # the same id/data self.assertMessageEqual(last_msg, next_last_msg) # Check the message id/data sent is the same as message received # Set timestamp and channel to match recv'd because we don't care # and they are not initialized by the can.Message constructor. msg.timestamp = last_msg.timestamp msg.channel = last_msg.channel self.assertMessageEqual(msg, last_msg) def test_removing_bus_tasks(self): bus = can.interface.Bus(interface="virtual") tasks = [] for task_i in range(10): msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7], ) msg.arbitration_id = task_i task = bus.send_periodic(msg, 0.1, 1) tasks.append(task) self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC) assert len(bus._periodic_tasks) == 10 for task in tasks: # Note calling task.stop will remove the task from the Bus's internal task management list task.stop() assert len(bus._periodic_tasks) == 0 bus.shutdown() def test_managed_tasks(self): bus = can.interface.Bus(interface="virtual", receive_own_messages=True) tasks = [] for task_i in range(3): msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7], ) msg.arbitration_id = task_i task = bus.send_periodic(msg, 0.1, 10, store_task=False) tasks.append(task) self.assertIsInstance(task, can.broadcastmanager.CyclicSendTaskABC) assert len(bus._periodic_tasks) == 0 # Self managed tasks should still be sending messages for _ in range(50): received_msg = bus.recv(timeout=5.0) assert received_msg is not None assert received_msg.arbitration_id in {0, 1, 2} for task in tasks: task.stop() for task in tasks: assert task.thread.join(5.0) is None, "Task didn't stop before timeout" bus.shutdown() def test_stopping_perodic_tasks(self): bus = can.interface.Bus(interface="virtual") tasks = [] for task_i in range(10): msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7], ) msg.arbitration_id = task_i task = bus.send_periodic(msg, 0.1, 1) tasks.append(task) assert len(bus._periodic_tasks) == 10 # stop half the tasks using the task object for task in tasks[::2]: task.stop() assert len(bus._periodic_tasks) == 5 # stop the other half using the bus api bus.stop_all_periodic_tasks(remove_tasks=False) for task in tasks: assert task.thread.join(5.0) is None, "Task didn't stop before timeout" # Tasks stopped via `stop_all_periodic_tasks` with remove_tasks=False should # still be associated with the bus (e.g. for restarting) assert len(bus._periodic_tasks) == 5 bus.shutdown() def test_restart_perodic_tasks(self): period = 0.01 safe_timeout = period * 5 msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7] ) with can.ThreadSafeBus(interface="virtual", receive_own_messages=True) as bus: task = bus.send_periodic(msg, period) self.assertIsInstance(task, can.broadcastmanager.RestartableCyclicTaskABC) # Test that the task is sending messages sleep(safe_timeout) assert not bus.queue.empty(), "messages should have been transmitted" # Stop the task and check that messages are no longer being sent bus.stop_all_periodic_tasks(remove_tasks=False) sleep(safe_timeout) while not bus.queue.empty(): bus.recv(timeout=period) sleep(safe_timeout) assert bus.queue.empty(), "messages should not have been transmitted" # Restart the task and check that messages are being sent again task.start() sleep(safe_timeout) assert not bus.queue.empty(), "messages should have been transmitted" # Stop all tasks and wait for the thread to exit bus.stop_all_periodic_tasks() if isinstance(task, can.broadcastmanager.ThreadBasedCyclicSendTask): # Avoids issues where the thread is still running when the bus is shutdown task.thread.join(safe_timeout) @unittest.skipIf(IS_CI, "fails randomly when run on CI server") def test_thread_based_cyclic_send_task(self): bus = can.ThreadSafeBus(interface="virtual") msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7] ) # good case, bus is up on_error_mock = MagicMock(return_value=False) task = can.broadcastmanager.ThreadBasedCyclicSendTask( bus=bus, lock=bus._lock_send_periodic, messages=msg, period=0.1, duration=3, on_error=on_error_mock, ) sleep(1) on_error_mock.assert_not_called() task.stop() bus.shutdown() # bus has been shut down on_error_mock = MagicMock(return_value=False) task = can.broadcastmanager.ThreadBasedCyclicSendTask( bus=bus, lock=bus._lock_send_periodic, messages=msg, period=0.1, duration=3, on_error=on_error_mock, ) sleep(1) self.assertEqual(1, on_error_mock.call_count) task.stop() # bus is still shut down, but on_error returns True on_error_mock = MagicMock(return_value=True) task = can.broadcastmanager.ThreadBasedCyclicSendTask( bus=bus, lock=bus._lock_send_periodic, messages=msg, period=0.1, duration=3, on_error=on_error_mock, ) sleep(1) self.assertTrue(on_error_mock.call_count > 1) task.stop() def test_modifier_callback(self) -> None: msg_list: List[can.Message] = [] def increment_first_byte(msg: can.Message) -> None: msg.data[0] += 1 original_msg = can.Message( is_extended_id=False, arbitration_id=0x123, data=[0] * 8 ) with can.ThreadSafeBus(interface="virtual", receive_own_messages=True) as bus: notifier = can.Notifier(bus=bus, listeners=[msg_list.append]) task = bus.send_periodic( msgs=original_msg, period=0.001, modifier_callback=increment_first_byte ) time.sleep(0.2) task.stop() notifier.stop() self.assertEqual(b"\x01\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[0].data)) self.assertEqual(b"\x02\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[1].data)) self.assertEqual(b"\x03\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[2].data)) self.assertEqual(b"\x04\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[3].data)) self.assertEqual(b"\x05\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[4].data)) self.assertEqual(b"\x06\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[5].data)) self.assertEqual(b"\x07\x00\x00\x00\x00\x00\x00\x00", bytes(msg_list[6].data)) if __name__ == "__main__": unittest.main()