@@ -0,0 +1,126 @@ #!/usr/bin/env python
""" This module tests the functions inside of bridge.py """
import random import string import sys import threading import time from time import sleep as real_sleep import unittest.mock
import can import can.bridge from can.interfaces import virtual
from .message_helper import ComparingMessagesTestCase

class TestBridgeScriptModule(unittest.TestCase, ComparingMessagesTestCase):
TIMEOUT = 3.0
def __init__(self, *args, **kwargs): unittest.TestCase.__init__(self, *args, **kwargs) ComparingMessagesTestCase.__init__( self, allowed_timestamp_delta=None, preserves_channel=False, )
def setUp(self) -> None: self.stop_event = threading.Event()
self.channel1 = "".join(random.choices(string.ascii_letters, k=8)) self.channel2 = "".join(random.choices(string.ascii_letters, k=8))
self.cli_args = [ "--bus1-interface", "virtual", "--bus1-channel", self.channel1, "--bus2-interface", "virtual", "--bus2-channel", self.channel2, ]
self.testmsg = can.Message( arbitration_id=0xC0FFEE, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=True )
def fake_sleep(self, duration): """A fake replacement for time.sleep that checks periodically whether self.stop_event is set, and raises KeyboardInterrupt if so.
This allows tests to simulate an interrupt (like Ctrl+C) during long sleeps, in a controlled and responsive way. """ interval = 0.05 # Small interval for responsiveness t_wakeup = time.perf_counter() + duration while time.perf_counter() < t_wakeup: if self.stop_event.is_set(): raise KeyboardInterrupt("Simulated interrupt from fake_sleep") real_sleep(interval)
def test_bridge(self): with ( unittest.mock.patch("can.bridge.time.sleep", new=self.fake_sleep), unittest.mock.patch("can.bridge.sys.argv", [sys.argv[0], *self.cli_args]), ): # start script thread = threading.Thread(target=can.bridge.main) thread.start()
# wait until script instantiates virtual buses t0 = time.perf_counter() while True: with virtual.channels_lock: if ( self.channel1 in virtual.channels and self.channel2 in virtual.channels ): break if time.perf_counter() > t0 + 2.0: raise TimeoutError("Bridge script did not create virtual buses") real_sleep(0.2)
# create buses with the same channels as in scripts with ( can.interfaces.virtual.VirtualBus(self.channel1) as bus1, can.interfaces.virtual.VirtualBus(self.channel2) as bus2, ): # send test message to bus1, it should be received on bus2 bus1.send(self.testmsg) recv_msg = bus2.recv(self.TIMEOUT) self.assertMessageEqual(self.testmsg, recv_msg)
# assert that both buses are empty self.assertIsNone(bus1.recv(0)) self.assertIsNone(bus2.recv(0))
# send test message to bus2, it should be received on bus1 bus2.send(self.testmsg) recv_msg = bus1.recv(self.TIMEOUT) self.assertMessageEqual(self.testmsg, recv_msg)
# assert that both buses are empty self.assertIsNone(bus1.recv(0)) self.assertIsNone(bus2.recv(0))
# stop the bridge script self.stop_event.set() thread.join()
# assert that the virtual buses were closed with virtual.channels_lock: self.assertNotIn(self.channel1, virtual.channels) self.assertNotIn(self.channel2, virtual.channels)

if __name__ == "__main__": unittest.main()