Skip to content

Commit 2a73b8e

Browse files
zariiii9003zariiii9003
committed
use new api to create cli args for bus1 and bus2
1 parent 9f0950e commit 2a73b8e

File tree

3 files changed

+136
-179
lines changed

3 files changed

+136
-179
lines changed

‎can/bridge.py‎

Lines changed: 33 additions & 138 deletions
Original file line numberDiff line numberDiff line change
@@ -7,160 +7,55 @@
77

88
import argparse
99
import errno
10-
import logging
1110
import sys
1211
import time
1312
from datetime import datetime
14-
from typing import (
15-
Iterator,
16-
List,
17-
Tuple,
18-
)
13+
from typing import Final
1914

20-
import can
21-
22-
from .logger import _create_base_argument_parser, _create_bus, _parse_additional_config
23-
24-
USAGE = """
25-
usage: can_bridge [{general config} --] {can A config} -- {can B config}
15+
from can.cli import add_bus_arguments, create_bus_from_namespace
16+
from can.listener import RedirectReader
17+
from can.notifier import Notifier
2618

19+
BRIDGE_DESCRIPTION: Final = """\
2720
Bridge two CAN buses.
2821
29-
Both can buses will be connected so that messages from bus A will be sent on
30-
bus B and messages on bus B will be sent to bus A. The buses are separated by a `--`
31-
32-
positional arguments:
33-
{general config} The configuration for this program excluding
34-
the config for each bus. Can be omitted
35-
{can A config} The configuration for the first bus
36-
{can B config} The configuration for the second bus
37-
38-
Example usage:
39-
can_bridge -i socketcan -c can0 -- -i socketcan can1
40-
can_bridge -vvv -- -i socketcan -c can0 -- -i socketcan can1
41-
42-
Type `can_bridge help_bus` for information about single bus configuration.
22+
Both can buses will be connected so that messages from bus1 will be sent on
23+
bus2 and messages from bus2 will be sent to bus1.
4324
"""
44-
45-
LOG = logging.getLogger(__name__)
46-
47-
48-
class UserError(Exception):
49-
pass
50-
51-
52-
def get_config_list(it: Iterator[str], separator: str, conf: list) -> None:
53-
while True:
54-
el = next(it)
55-
if el == separator:
56-
break
57-
58-
conf.append(el)
25+
BUS_1_PREFIX: Final = "bus1"
26+
BUS_2_PREFIX: Final = "bus2"
5927

6028

61-
def split_configurations(
62-
arg_list: List[str], separator: str = "--"
63-
) -> Tuple[list, list, list]:
64-
general = []
65-
conf_a: List[str] = []
66-
conf_b: List[str] = []
29+
def _parse_bridge_args(args: list[str]) -> argparse.Namespace:
30+
"""Parse command line arguments for bridge script."""
6731

68-
found_sep = False
69-
it = iter(arg_list)
70-
try:
71-
get_config_list(it, separator, conf_a)
72-
found_sep = True
73-
get_config_list(it, separator, conf_b)
32+
parser = argparse.ArgumentParser(description=BRIDGE_DESCRIPTION)
33+
add_bus_arguments(parser, prefix=BUS_1_PREFIX, group_title="Bus 1 arguments")
34+
add_bus_arguments(parser, prefix=BUS_2_PREFIX, group_title="Bus 2 arguments")
7435

75-
# When we reached this point we found two separators so we have
76-
# a general config. We will treate the first config as general
77-
general = conf_a
78-
conf_a = conf_b
79-
get_config_list(it, separator, conf_b)
80-
81-
# When we reached this point we found three separators so this is
82-
# an error.
83-
raise UserError("To many configurations")
84-
except StopIteration:
85-
LOG.debug("All configurations were split")
86-
if not found_sep:
87-
raise UserError("Missing separator") from None
36+
# print help message when no arguments were given
37+
if not args:
38+
parser.print_help(sys.stderr)
39+
raise SystemExit(errno.EINVAL)
8840

89-
return general, conf_a, conf_b
41+
results, _unknown_args = parser.parse_known_args(args)
42+
return results
9043

9144

9245
def main() -> None:
93-
general_parser = argparse.ArgumentParser()
94-
general_parser.add_argument(
95-
"-v",
96-
action="count",
97-
dest="verbosity",
98-
help="""How much information do you want to see at the command line?
99-
You can add several of these e.g., -vv is DEBUG""",
100-
default=2,
101-
)
102-
103-
bus_parser = argparse.ArgumentParser(description="Bridge two CAN buses.")
104-
105-
_create_base_argument_parser(bus_parser)
106-
107-
parser = argparse.ArgumentParser(description="Bridge two CAN buses.")
108-
parser.add_argument("configs", nargs=argparse.REMAINDER)
109-
110-
# print help message when no arguments were given
111-
if len(sys.argv) < 2:
112-
print(USAGE, file=sys.stderr)
113-
raise SystemExit(errno.EINVAL)
114-
115-
args = sys.argv[1:]
116-
try:
117-
general, conf_a, conf_b = split_configurations(args)
118-
except UserError as exc:
119-
if len(args) >= 1:
120-
if args[0] == "-h" or args[0] == "--help" or args[0] == "help":
121-
print(USAGE)
122-
raise SystemExit() from None
123-
elif args[0] == "help_bus":
124-
bus_parser.print_help(sys.stderr)
125-
else:
126-
print(f"Error while processing arguments: {exc}", file=sys.stderr)
127-
raise SystemExit(errno.EINVAL) from exc
128-
129-
LOG.debug("General configuration: %s", general)
130-
LOG.debug("Bus A configuration: %s", conf_a)
131-
LOG.debug("Bus B configuration: %s", conf_b)
132-
g_results = general_parser.parse_args(general)
133-
verbosity = g_results.verbosity
134-
135-
a_results, a_unknown_args = bus_parser.parse_known_args(conf_a)
136-
a_additional_config = _parse_additional_config(
137-
[*a_results.extra_args, *a_unknown_args]
138-
)
139-
a_results.__dict__["verbosity"] = verbosity
140-
141-
b_results, b_unknown_args = bus_parser.parse_known_args(conf_b)
142-
b_additional_config = _parse_additional_config(
143-
[*b_results.extra_args, *b_unknown_args]
144-
)
145-
b_results.__dict__["verbosity"] = verbosity
146-
147-
LOG.debug("General configuration results: %s", g_results)
148-
LOG.debug("Bus A configuration results: %s", a_results)
149-
LOG.debug("Bus A additional configuration results: %s", a_additional_config)
150-
LOG.debug("Bus B configuration results: %s", b_results)
151-
LOG.debug("Bus B additional configuration results: %s", b_additional_config)
152-
with _create_bus(a_results, **a_additional_config) as bus_a:
153-
with _create_bus(b_results, **b_additional_config) as bus_b:
154-
reader_a = can.RedirectReader(bus_b)
155-
reader_b = can.RedirectReader(bus_a)
156-
can.Notifier(bus_a, [reader_a])
157-
can.Notifier(bus_b, [reader_b])
158-
print(f"CAN Bridge (Started on {datetime.now()})")
159-
try:
160-
while True:
161-
time.sleep(1)
162-
except KeyboardInterrupt:
163-
pass
46+
results = _parse_bridge_args(sys.argv[1:])
47+
48+
with create_bus_from_namespace(results, prefix=BUS_1_PREFIX) as bus1:
49+
with create_bus_from_namespace(results, prefix=BUS_2_PREFIX) as bus2:
50+
reader1_to_2 = RedirectReader(bus2)
51+
reader2_to_1 = RedirectReader(bus1)
52+
with Notifier(bus1, [reader1_to_2]), Notifier(bus2, [reader2_to_1]):
53+
print(f"CAN Bridge (Started on {datetime.now()})")
54+
try:
55+
while True:
56+
time.sleep(1)
57+
except KeyboardInterrupt:
58+
pass
16459

16560
print(f"CAN Bridge (Stopped on {datetime.now()})")
16661

‎doc/scripts.rst‎

Lines changed: 0 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -66,12 +66,6 @@ A small application that can be used to connect two can buses:
6666
:shell:
6767

6868

69-
Example call:
70-
::
71-
72-
python -m can.bridge -i socketcan -c can0 -- -i socketcan -c can1
73-
74-
7569
can.logconvert
7670
--------------
7771

‎test/test_bridge.py‎

Lines changed: 103 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -4,54 +4,122 @@
44
This module tests the functions inside of bridge.py
55
"""
66

7+
import random
8+
import string
79
import sys
8-
import unittest
9-
from unittest import mock
10-
from unittest.mock import Mock
10+
import threading
11+
import time
12+
from time import sleep as real_sleep
13+
import unittest.mock
1114

1215
import can
1316
import can.bridge
17+
from can.interfaces import virtual
1418

19+
from .message_helper import ComparingMessagesTestCase
20+
21+
22+
class TestBridgeScriptModule(unittest.TestCase, ComparingMessagesTestCase):
23+
24+
TIMEOUT = 3.0
25+
26+
def __init__(self, *args, **kwargs):
27+
unittest.TestCase.__init__(self, *args, **kwargs)
28+
ComparingMessagesTestCase.__init__(
29+
self,
30+
allowed_timestamp_delta=None,
31+
preserves_channel=False,
32+
)
1533

16-
class TestBridgeScriptModule(unittest.TestCase):
1734
def setUp(self) -> None:
18-
# Patch VirtualBus object
19-
patcher_virtual_bus = mock.patch("can.interfaces.virtual.VirtualBus", spec=True)
20-
self.MockVirtualBus = patcher_virtual_bus.start()
21-
self.addCleanup(patcher_virtual_bus.stop)
22-
self.mock_virtual_bus = self.MockVirtualBus.return_value
23-
self.mock_virtual_bus.__enter__ = Mock(return_value=self.mock_virtual_bus)
24-
25-
# Patch time sleep object
26-
patcher_sleep = mock.patch("can.bridge.time.sleep", spec=True)
27-
self.MockSleep = patcher_sleep.start()
28-
self.addCleanup(patcher_sleep.stop)
35+
self.stop_event = threading.Event()
36+
37+
self.channel1 = "".join(random.choices(string.ascii_letters, k=8))
38+
self.channel2 = "".join(random.choices(string.ascii_letters, k=8))
39+
40+
self.cli_args = [
41+
"--bus1-interface",
42+
"virtual",
43+
"--bus1-channel",
44+
self.channel1,
45+
"--bus2-interface",
46+
"virtual",
47+
"--bus2-channel",
48+
self.channel2,
49+
]
2950

3051
self.testmsg = can.Message(
3152
arbitration_id=0xC0FFEE, data=[0, 25, 0, 1, 3, 1, 4, 1], is_extended_id=True
3253
)
3354

34-
self.busargs = ["-i", "virtual"]
35-
36-
def assertSuccessfullCleanup(self):
37-
self.MockVirtualBus.assert_called()
38-
self.assertEqual(2, len(self.mock_virtual_bus.__exit__.mock_calls))
39-
40-
def test_bridge_no_config(self):
41-
self.MockSleep.side_effect = KeyboardInterrupt
42-
sys.argv = [
43-
sys.argv[0],
44-
*self.busargs,
45-
"-c",
46-
"can_a",
47-
"--",
48-
*self.busargs,
49-
"-c",
50-
"can_b",
51-
]
52-
can.bridge.main()
55+
def fake_sleep(self, duration):
56+
"""A fake replacement for time.sleep that checks periodically
57+
whether self.stop_event is set, and raises KeyboardInterrupt
58+
if so.
59+
60+
This allows tests to simulate an interrupt (like Ctrl+C)
61+
during long sleeps, in a controlled and responsive way.
62+
"""
63+
interval = 0.05 # Small interval for responsiveness
64+
t_wakeup = time.perf_counter() + duration
65+
while time.perf_counter() < t_wakeup:
66+
if self.stop_event.is_set():
67+
raise KeyboardInterrupt("Simulated interrupt from fake_sleep")
68+
real_sleep(interval)
69+
70+
def test_bridge(self):
71+
with (
72+
unittest.mock.patch("can.bridge.time.sleep", new=self.fake_sleep),
73+
unittest.mock.patch("can.bridge.sys.argv", [sys.argv[0], *self.cli_args]),
74+
):
75+
# start script
76+
thread = threading.Thread(target=can.bridge.main)
77+
thread.start()
78+
79+
# wait until script instantiates virtual buses
80+
t0 = time.perf_counter()
81+
while True:
82+
with virtual.channels_lock:
83+
if (
84+
self.channel1 in virtual.channels
85+
and self.channel2 in virtual.channels
86+
):
87+
break
88+
if time.perf_counter() > t0 + 2.0:
89+
raise TimeoutError("Bridge script did not create virtual buses")
90+
real_sleep(0.2)
91+
92+
# create buses with the same channels as in scripts
93+
with (
94+
can.interfaces.virtual.VirtualBus(self.channel1) as bus1,
95+
can.interfaces.virtual.VirtualBus(self.channel2) as bus2,
96+
):
97+
# send test message to bus1, it should be received on bus2
98+
bus1.send(self.testmsg)
99+
recv_msg = bus2.recv(self.TIMEOUT)
100+
self.assertMessageEqual(self.testmsg, recv_msg)
101+
102+
# assert that both buses are empty
103+
self.assertIsNone(bus1.recv(0))
104+
self.assertIsNone(bus2.recv(0))
105+
106+
# send test message to bus2, it should be received on bus1
107+
bus2.send(self.testmsg)
108+
recv_msg = bus1.recv(self.TIMEOUT)
109+
self.assertMessageEqual(self.testmsg, recv_msg)
110+
111+
# assert that both buses are empty
112+
self.assertIsNone(bus1.recv(0))
113+
self.assertIsNone(bus2.recv(0))
114+
115+
# stop the bridge script
116+
self.stop_event.set()
117+
thread.join()
53118

54-
self.assertSuccessfullCleanup()
119+
# assert that the virtual buses were closed
120+
with virtual.channels_lock:
121+
self.assertNotIn(self.channel1, virtual.channels)
122+
self.assertNotIn(self.channel2, virtual.channels)
55123

56124

57125
if __name__ == "__main__":

0 commit comments

Comments
 (0)