From 33a1ec740137345075390fe5252db69e27668294 Mon Sep 17 00:00:00 2001 From: zariiii9003 <52598363+zariiii9003@users.noreply.github.com> Date: Sat, 23 Nov 2024 14:39:43 +0100 Subject: [PATCH] Fix CI failures (#1898) * implement join_threads helper method * simplify network_test.py * update tox.ini --- test/network_test.py | 103 +++++++++++++++----------------------- test/simplecyclic_test.py | 40 +++++++++++---- tox.ini | 4 +- 3 files changed, 71 insertions(+), 76 deletions(-) diff --git a/test/network_test.py b/test/network_test.py index 50070ef40..3a231dff7 100644 --- a/test/network_test.py +++ b/test/network_test.py @@ -6,16 +6,14 @@ import unittest import can +from test.config import IS_PYPY logging.getLogger(__file__).setLevel(logging.WARNING) # make a random bool: def rbool(): - return bool(round(random.random())) - - -channel = "vcan0" + return random.choice([False, True]) class ControllerAreaNetworkTestCase(unittest.TestCase): @@ -51,74 +49,51 @@ def tearDown(self): # Restore the defaults can.rc = self._can_rc - def producer(self, ready_event, msg_read): - self.client_bus = can.interface.Bus(channel=channel) - ready_event.wait() - for i in range(self.num_messages): - m = can.Message( - arbitration_id=self.ids[i], - is_remote_frame=self.remote_flags[i], - is_error_frame=self.error_flags[i], - is_extended_id=self.extended_flags[i], - data=self.data[i], - ) - # logging.debug("writing message: {}".format(m)) - if msg_read is not None: - # Don't send until the other thread is ready - msg_read.wait() - msg_read.clear() - - self.client_bus.send(m) + def producer(self, channel: str): + with can.interface.Bus(channel=channel) as client_bus: + for i in range(self.num_messages): + m = can.Message( + arbitration_id=self.ids[i], + is_remote_frame=self.remote_flags[i], + is_error_frame=self.error_flags[i], + is_extended_id=self.extended_flags[i], + data=self.data[i], + ) + client_bus.send(m) def testProducer(self): """Verify that we can send arbitrary messages on the bus""" logging.debug("testing producer alone") - ready = threading.Event() - ready.set() - self.producer(ready, None) - + self.producer(channel="testProducer") logging.debug("producer test complete") def testProducerConsumer(self): logging.debug("testing producer/consumer") - ready = threading.Event() - msg_read = threading.Event() - - self.server_bus = can.interface.Bus(channel=channel, interface="virtual") - - t = threading.Thread(target=self.producer, args=(ready, msg_read)) - t.start() - - # Ensure there are no messages on the bus - while True: - m = self.server_bus.recv(timeout=0.5) - if m is None: - print("No messages... lets go") - break - else: - self.fail("received messages before the test has started ...") - ready.set() - i = 0 - while i < self.num_messages: - msg_read.set() - msg = self.server_bus.recv(timeout=0.5) - self.assertIsNotNone(msg, "Didn't receive a message") - # logging.debug("Received message {} with data: {}".format(i, msg.data)) - - self.assertEqual(msg.is_extended_id, self.extended_flags[i]) - if not msg.is_remote_frame: - self.assertEqual(msg.data, self.data[i]) - self.assertEqual(msg.arbitration_id, self.ids[i]) - - self.assertEqual(msg.is_error_frame, self.error_flags[i]) - self.assertEqual(msg.is_remote_frame, self.remote_flags[i]) - - i += 1 - t.join() - - with contextlib.suppress(NotImplementedError): - self.server_bus.flush_tx_buffer() - self.server_bus.shutdown() + read_timeout = 2.0 if IS_PYPY else 0.5 + channel = "testProducerConsumer" + + with can.interface.Bus(channel=channel, interface="virtual") as server_bus: + t = threading.Thread(target=self.producer, args=(channel,)) + t.start() + + i = 0 + while i < self.num_messages: + msg = server_bus.recv(timeout=read_timeout) + self.assertIsNotNone(msg, "Didn't receive a message") + + self.assertEqual(msg.is_extended_id, self.extended_flags[i]) + if not msg.is_remote_frame: + self.assertEqual(msg.data, self.data[i]) + self.assertEqual(msg.arbitration_id, self.ids[i]) + + self.assertEqual(msg.is_error_frame, self.error_flags[i]) + self.assertEqual(msg.is_remote_frame, self.remote_flags[i]) + + i += 1 + t.join() + + with contextlib.suppress(NotImplementedError): + server_bus.flush_tx_buffer() if __name__ == "__main__": diff --git a/test/simplecyclic_test.py b/test/simplecyclic_test.py index 7116efc9d..c4c1a2340 100644 --- a/test/simplecyclic_test.py +++ b/test/simplecyclic_test.py @@ -5,8 +5,11 @@ """ import gc +import sys import time +import traceback import unittest +from threading import Thread from time import sleep from typing import List from unittest.mock import MagicMock @@ -87,6 +90,8 @@ def test_removing_bus_tasks(self): # Note calling task.stop will remove the task from the Bus's internal task management list task.stop() + self.join_threads([task.thread for task in tasks], 5.0) + assert len(bus._periodic_tasks) == 0 bus.shutdown() @@ -115,8 +120,7 @@ def test_managed_tasks(self): for task in tasks: task.stop() - for task in tasks: - assert task.thread.join(5.0) is None, "Task didn't stop before timeout" + self.join_threads([task.thread for task in tasks], 5.0) bus.shutdown() @@ -142,9 +146,7 @@ def test_stopping_perodic_tasks(self): # 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" + self.join_threads([task.thread for task in tasks], 5.0) # Tasks stopped via `stop_all_periodic_tasks` with remove_tasks=False should # still be associated with the bus (e.g. for restarting) @@ -161,7 +163,7 @@ def test_restart_perodic_tasks(self): is_extended_id=False, arbitration_id=0x123, data=[0, 1, 2, 3, 4, 5, 6, 7] ) - def _read_all_messages(_bus: can.interfaces.virtual.VirtualBus) -> None: + def _read_all_messages(_bus: "can.interfaces.virtual.VirtualBus") -> None: sleep(safe_timeout) while not _bus.queue.empty(): _bus.recv(timeout=period) @@ -207,9 +209,8 @@ def _read_all_messages(_bus: can.interfaces.virtual.VirtualBus) -> None: # 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) + # Avoids issues where the thread is still running when the bus is shutdown + self.join_threads([task.thread], 5.0) @unittest.skipIf(IS_CI, "fails randomly when run on CI server") def test_thread_based_cyclic_send_task(self): @@ -288,6 +289,27 @@ def increment_first_byte(msg: can.Message) -> None: 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)) + @staticmethod + def join_threads(threads: List[Thread], timeout: float) -> None: + stuck_threads: List[Thread] = [] + t0 = time.perf_counter() + for thread in threads: + time_left = timeout - (time.perf_counter() - t0) + if time_left > 0.0: + thread.join(time_left) + if thread.is_alive(): + if platform.python_implementation() == "CPython": + # print thread frame to help with debugging + frame = sys._current_frames()[thread.ident] + traceback.print_stack(frame, file=sys.stderr) + stuck_threads.append(thread) + if stuck_threads: + err_message = ( + f"Threads did not stop within {timeout:.1f} seconds: " + f"[{', '.join([str(t) for t in stuck_threads])}]" + ) + raise RuntimeError(err_message) + if __name__ == "__main__": unittest.main() diff --git a/tox.ini b/tox.ini index 1d4e88166..e112c22b4 100644 --- a/tox.ini +++ b/tox.ini @@ -11,7 +11,7 @@ deps = pyserial~=3.5 parameterized~=0.8 asammdf>=6.0; platform_python_implementation=="CPython" and python_version<"3.13" - pywin32>=305; platform_system=="Windows" and platform_python_implementation=="CPython" and python_version<"3.13" + pywin32>=305; platform_system=="Windows" and platform_python_implementation=="CPython" and python_version<"3.14" commands = pytest {posargs} @@ -19,8 +19,6 @@ commands = extras = canalystii -recreate = True - [testenv:gh] passenv = CI