Skip to content

Commit 14b3336

Browse files
friederschueleracolombbizfsc
authored
emcy: Catch callback exceptions and expand test coverage (#659)
- Fix bug where an exception in one EMCY callback would abort all subsequent callbacks; now logs the exception and continues - Add test for callback exception handling (verifies fix) - Add tests for EmcyConsumer initialization, multiple callback order, error reset variants, wait timeout edge cases, concurrent errors - Add tests for EmcyError initialization types and str representation - Add tests for EmcyProducer initialization, send/reset edge cases - Add TestEmcyIntegration class using network.subscribe() for producer-consumer roundtrip tests - Remove superfluous inline comments per reviewer feedback (acolomb) - Convert leading comment to docstring in test_emcy_consumer_on_emcy Addresses review feedback from PR #604 by yuvraj-kolkar17. * Factor out mock_rx_thread(). Simplify by extracting Thread creation and starting, and joining to one central helper function, including the emcy_received condition lock to avoid race conditions. Co-authored-by: André Colomb <src@andre.colomb.de> Co-authored-by: Frieder Schüler <frieder.schueler@bizerba.com>
1 parent 3b57fed commit 14b3336

2 files changed

Lines changed: 205 additions & 64 deletions

File tree

‎canopen/emcy.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,10 @@ def on_emcy(self, can_id, data, timestamp):
3939
self.emcy_received.notify_all()
4040

4141
for callback in self.callbacks:
42-
callback(entry)
42+
try:
43+
callback(entry)
44+
except Exception:
45+
logger.exception("Exception in EMCY callback")
4346

4447
def add_callback(self, callback: Callable[[EmcyError], None]):
4548
"""Get notified on EMCY messages from this node.

‎test/test_emcy.py‎

Lines changed: 201 additions & 63 deletions
Original file line numberDiff line numberDiff line change
@@ -6,78 +6,85 @@
66
import can
77

88
import canopen
9-
from canopen.emcy import EmcyError
109

1110

1211
TIMEOUT = 0.1
1312

1413

14+
@contextmanager
15+
def mock_rx_thread(consumer: canopen.emcy.EmcyConsumer, func):
16+
t = threading.Thread(target=func)
17+
try:
18+
with consumer.emcy_received:
19+
t.start()
20+
yield t
21+
finally:
22+
t.join(TIMEOUT)
23+
24+
1525
class TestEmcy(unittest.TestCase):
16-
def setUp(self):
17-
self.emcy = canopen.emcy.EmcyConsumer()
1826

1927
def check_error(self, err, code, reg, data, ts):
20-
self.assertIsInstance(err, EmcyError)
28+
self.assertIsInstance(err, canopen.emcy.EmcyError)
2129
self.assertIsInstance(err, Exception)
2230
self.assertEqual(err.code, code)
2331
self.assertEqual(err.register, reg)
2432
self.assertEqual(err.data, data)
2533
self.assertAlmostEqual(err.timestamp, ts)
2634

2735
def test_emcy_consumer_on_emcy(self):
28-
# Make sure multiple callbacks receive the same information.
36+
"""Make sure multiple callbacks receive the same information."""
37+
emcy = canopen.emcy.EmcyConsumer()
2938
acc1 = []
3039
acc2 = []
31-
self.emcy.add_callback(lambda err: acc1.append(err))
32-
self.emcy.add_callback(lambda err: acc2.append(err))
40+
emcy.add_callback(lambda err: acc1.append(err))
41+
emcy.add_callback(lambda err: acc2.append(err))
3342

34-
# Dispatch an EMCY datagram.
35-
self.emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
43+
emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
3644

37-
self.assertEqual(len(self.emcy.log), 1)
38-
self.assertEqual(len(self.emcy.active), 1)
45+
self.assertEqual(len(emcy.log), 1)
46+
self.assertEqual(len(emcy.active), 1)
3947

40-
error = self.emcy.log[0]
41-
self.assertEqual(self.emcy.active[0], error)
48+
error = emcy.log[0]
49+
self.assertEqual(emcy.active[0], error)
4250
for err in error, acc1[0], acc2[0]:
4351
self.check_error(
4452
error, code=0x2001, reg=0x02,
4553
data=bytes([0, 1, 2, 3, 4]), ts=1000,
4654
)
4755

48-
# Dispatch a new EMCY datagram.
49-
self.emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000)
50-
self.assertEqual(len(self.emcy.log), 2)
51-
self.assertEqual(len(self.emcy.active), 2)
56+
emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000)
57+
self.assertEqual(len(emcy.log), 2)
58+
self.assertEqual(len(emcy.active), 2)
5259

53-
error = self.emcy.log[1]
54-
self.assertEqual(self.emcy.active[1], error)
60+
error = emcy.log[1]
61+
self.assertEqual(emcy.active[1], error)
5562
for err in error, acc1[1], acc2[1]:
5663
self.check_error(
5764
error, code=0x9010, reg=0x01,
5865
data=bytes([4, 3, 2, 1, 0]), ts=2000,
5966
)
6067

61-
# Dispatch an EMCY reset.
62-
self.emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 2000)
63-
self.assertEqual(len(self.emcy.log), 3)
64-
self.assertEqual(len(self.emcy.active), 0)
68+
emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 2000)
69+
self.assertEqual(len(emcy.log), 3)
70+
self.assertEqual(len(emcy.active), 0)
6571

6672
def test_emcy_consumer_reset(self):
67-
self.emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
68-
self.emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000)
69-
self.assertEqual(len(self.emcy.log), 2)
70-
self.assertEqual(len(self.emcy.active), 2)
73+
emcy = canopen.emcy.EmcyConsumer()
74+
emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
75+
emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000)
76+
self.assertEqual(len(emcy.log), 2)
77+
self.assertEqual(len(emcy.active), 2)
7178

72-
self.emcy.reset()
73-
self.assertEqual(len(self.emcy.log), 0)
74-
self.assertEqual(len(self.emcy.active), 0)
79+
emcy.reset()
80+
self.assertEqual(len(emcy.log), 0)
81+
self.assertEqual(len(emcy.active), 0)
7582

7683
def test_emcy_consumer_wait(self):
77-
PAUSE = TIMEOUT / 2
84+
emcy = canopen.emcy.EmcyConsumer()
7885

7986
def push_err():
80-
self.emcy.on_emcy(0x81, b'\x01\x20\x01\x01\x02\x03\x04\x05', 100)
87+
emcy.on_emcy(0x81, b'\x01\x20\x01\x01\x02\x03\x04\x05', 100)
8188

8289
def check_err(err):
8390
self.assertIsNotNone(err)
@@ -86,55 +93,106 @@ def check_err(err):
8693
data=bytes([1, 2, 3, 4, 5]), ts=100,
8794
)
8895

89-
@contextmanager
90-
def timer(func):
91-
t = threading.Timer(PAUSE, func)
92-
try:
93-
yield t
94-
finally:
95-
t.join(TIMEOUT)
96-
9796
# Check unfiltered wait, on timeout.
98-
self.assertIsNone(self.emcy.wait(timeout=TIMEOUT))
97+
self.assertIsNone(emcy.wait(timeout=TIMEOUT))
9998

10099
# Check unfiltered wait, on success.
101-
with timer(push_err) as t:
102-
with self.assertLogs(level=logging.INFO):
103-
t.start()
104-
err = self.emcy.wait(timeout=TIMEOUT)
105-
check_err(err)
100+
with (
101+
self.assertLogs(level=logging.INFO),
102+
mock_rx_thread(emcy, push_err),
103+
):
104+
check_err(emcy.wait(timeout=TIMEOUT))
106105

107106
# Check filtered wait, on success.
108-
with timer(push_err) as t:
109-
with self.assertLogs(level=logging.INFO):
110-
t.start()
111-
err = self.emcy.wait(0x2001, TIMEOUT)
112-
check_err(err)
107+
with (
108+
self.assertLogs(level=logging.INFO),
109+
mock_rx_thread(emcy, push_err),
110+
):
111+
check_err(emcy.wait(0x2001, TIMEOUT))
113112

114113
# Check filtered wait, on timeout.
115-
with timer(push_err) as t:
116-
t.start()
117-
self.assertIsNone(self.emcy.wait(0x9000, TIMEOUT))
114+
with mock_rx_thread(emcy, push_err):
115+
self.assertIsNone(emcy.wait(0x9000, TIMEOUT))
118116

119117
def push_reset():
120-
self.emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 100)
121-
122-
with timer(push_reset) as t:
123-
t.start()
124-
self.assertIsNone(self.emcy.wait(0x9000, TIMEOUT))
118+
emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 100)
119+
120+
with mock_rx_thread(emcy, push_reset):
121+
self.assertIsNone(emcy.wait(0x9000, TIMEOUT))
122+
123+
def test_emcy_consumer_multiple_callbacks(self):
124+
"""Test adding multiple callbacks and their execution order."""
125+
emcy = canopen.emcy.EmcyConsumer()
126+
call_order = []
127+
emcy.add_callback(lambda err: call_order.append('callback1'))
128+
emcy.add_callback(lambda err: call_order.append('callback2'))
129+
emcy.add_callback(lambda err: call_order.append('callback3'))
130+
emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
131+
self.assertEqual(call_order, ['callback1', 'callback2', 'callback3'])
132+
133+
def test_emcy_consumer_callback_exception_handling(self):
134+
"""Test that callback exceptions don't break other callbacks or the system."""
135+
emcy = canopen.emcy.EmcyConsumer()
136+
successful_callbacks = []
137+
emcy.add_callback(lambda err: successful_callbacks.append('success1'))
138+
emcy.add_callback(
139+
lambda err: exec('raise ValueError("Test exception in callback")')
140+
)
141+
emcy.add_callback(lambda err: successful_callbacks.append('success2'))
142+
emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
143+
self.assertEqual(successful_callbacks, ['success1', 'success2'])
144+
145+
def test_emcy_consumer_error_reset_variants(self):
146+
"""Test different error reset code patterns."""
147+
emcy = canopen.emcy.EmcyConsumer()
148+
emcy.on_emcy(0x81, b'\x01\x20\x02\x00\x01\x02\x03\x04', 1000)
149+
emcy.on_emcy(0x81, b'\x10\x90\x01\x04\x03\x02\x01\x00', 2000)
150+
self.assertEqual(len(emcy.active), 2)
151+
emcy.on_emcy(0x81, b'\x00\x00\x00\x00\x00\x00\x00\x00', 3000)
152+
self.assertEqual(len(emcy.active), 0)
153+
emcy.on_emcy(0x81, b'\x01\x30\x02\x00\x01\x02\x03\x04', 4000)
154+
self.assertEqual(len(emcy.active), 1)
155+
emcy.on_emcy(0x81, b'\x99\x00\x01\x00\x00\x00\x00\x00', 5000)
156+
self.assertEqual(len(emcy.active), 0)
157+
158+
def test_emcy_consumer_wait_timeout_edge_cases(self):
159+
"""Test wait method with various timeout scenarios."""
160+
emcy = canopen.emcy.EmcyConsumer()
161+
result = emcy.wait(timeout=0)
162+
self.assertIsNone(result)
163+
result = emcy.wait(timeout=0.001)
164+
self.assertIsNone(result)
165+
166+
def test_emcy_consumer_wait_concurrent_errors(self):
167+
"""Test wait method when multiple errors arrive concurrently."""
168+
emcy = canopen.emcy.EmcyConsumer()
169+
170+
def push_multiple_errors():
171+
emcy.on_emcy(0x81, b'\x01\x20\x01\x01\x02\x03\x04\x05', 100)
172+
emcy.on_emcy(0x81, b'\x02\x20\x01\x01\x02\x03\x04\x05', 101)
173+
emcy.on_emcy(0x81, b'\x03\x20\x01\x01\x02\x03\x04\x05', 102)
174+
175+
with (
176+
self.assertLogs(level=logging.INFO),
177+
mock_rx_thread(emcy, push_multiple_errors),
178+
):
179+
err = emcy.wait(0x2003, timeout=TIMEOUT)
180+
self.assertIsNotNone(err)
181+
self.assertEqual(err.code, 0x2003)
125182

126183

127184
class TestEmcyError(unittest.TestCase):
185+
128186
def test_emcy_error(self):
129-
error = EmcyError(0x2001, 0x02, b'\x00\x01\x02\x03\x04', 1000)
187+
error = canopen.emcy.EmcyError(0x2001, 0x02, b'\x00\x01\x02\x03\x04', 1000)
130188
self.assertEqual(error.code, 0x2001)
131189
self.assertEqual(error.data, b'\x00\x01\x02\x03\x04')
132190
self.assertEqual(error.register, 2)
133191
self.assertEqual(error.timestamp, 1000)
134192

135193
def test_emcy_str(self):
136194
def check(code, expected):
137-
err = EmcyError(code, 1, b'', 1000)
195+
err = canopen.emcy.EmcyError(code, 1, b'', 1000)
138196
actual = str(err)
139197
self.assertEqual(actual, expected)
140198

@@ -145,7 +203,7 @@ def check(code, expected):
145203

146204
def test_emcy_get_desc(self):
147205
def check(code, expected):
148-
err = EmcyError(code, 1, b'', 1000)
206+
err = canopen.emcy.EmcyError(code, 1, b'', 1000)
149207
actual = err.get_desc()
150208
self.assertEqual(actual, expected)
151209

@@ -182,6 +240,7 @@ def check(code, expected):
182240

183241

184242
class TestEmcyProducer(unittest.TestCase):
243+
185244
def setUp(self):
186245
self.txbus = can.Bus(interface="virtual")
187246
self.rxbus = can.Bus(interface="virtual")
@@ -198,7 +257,7 @@ def tearDown(self):
198257

199258
def check_response(self, expected):
200259
msg = self.rxbus.recv(TIMEOUT)
201-
self.assertIsNotNone(msg)
260+
assert msg is not None
202261
actual = msg.data
203262
self.assertEqual(actual, expected)
204263

@@ -220,6 +279,85 @@ def check(*args, res):
220279
check(3, res=b'\x00\x00\x03\x00\x00\x00\x00\x00')
221280
check(3, b"\xaa\xbb", res=b'\x00\x00\x03\xaa\xbb\x00\x00\x00')
222281

282+
def test_emcy_producer_send_edge_cases(self):
283+
self.emcy.send(0xFFFF, 0xFF, b'\xFF\xFF\xFF\xFF\xFF')
284+
self.check_response(b'\xFF\xFF\xFF\xFF\xFF\xFF\xFF\xFF')
285+
self.emcy.send(0x0000, 0x00)
286+
self.check_response(b'\x00\x00\x00\x00\x00\x00\x00\x00')
287+
self.emcy.send(0x1234, 0x56, b'\xAB\xCD')
288+
self.check_response(b'\x34\x12\x56\xAB\xCD\x00\x00\x00')
289+
self.emcy.send(0x1234, 0x56, b'\xAB\xCD\xEF\x12\x34')
290+
self.check_response(b'\x34\x12\x56\xAB\xCD\xEF\x12\x34')
291+
292+
def test_emcy_producer_reset_edge_cases(self):
293+
self.emcy.reset(0xFF)
294+
self.check_response(b'\x00\x00\xFF\x00\x00\x00\x00\x00')
295+
self.emcy.reset(0xFF, b'\xFF\xFF\xFF\xFF\xFF')
296+
self.check_response(b'\x00\x00\xFF\xFF\xFF\xFF\xFF\xFF')
297+
self.emcy.reset(0x12, b'\xAB\xCD')
298+
self.check_response(b'\x00\x00\x12\xAB\xCD\x00\x00\x00')
299+
300+
301+
class TestEmcyIntegration(unittest.TestCase):
302+
"""Integration tests for EMCY producer and consumer."""
303+
304+
def setUp(self):
305+
self.txbus = can.Bus(interface="virtual")
306+
self.rxbus = can.Bus(interface="virtual")
307+
self.net = canopen.Network(self.txbus)
308+
self.net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
309+
self.net.connect()
310+
self.rx_net = canopen.Network(self.rxbus)
311+
self.rx_net.NOTIFIER_SHUTDOWN_TIMEOUT = 0.0
312+
self.rx_net.connect()
313+
self.producer = canopen.emcy.EmcyProducer(0x081)
314+
self.producer.network = self.net
315+
self.consumer = canopen.emcy.EmcyConsumer()
316+
self.rx_net.subscribe(0x081, self.consumer.on_emcy)
317+
318+
def tearDown(self):
319+
self.net.disconnect()
320+
self.rx_net.disconnect()
321+
self.txbus.shutdown()
322+
self.rxbus.shutdown()
323+
324+
def test_producer_consumer_integration(self):
325+
"""Test that producer and consumer work together."""
326+
received_errors = []
327+
self.consumer.add_callback(lambda err: received_errors.append(err))
328+
with (
329+
self.assertLogs(level=logging.INFO),
330+
mock_rx_thread(
331+
self.consumer,
332+
lambda: self.producer.send(0x2001, 0x02, b'\x01\x02\x03\x04\x05'),
333+
),
334+
):
335+
err = self.consumer.wait(0x2001, timeout=TIMEOUT)
336+
self.assertIsNotNone(err)
337+
self.assertEqual(err.code, 0x2001)
338+
self.assertEqual(err.register, 0x02)
339+
self.assertEqual(err.data, b'\x01\x02\x03\x04\x05')
340+
self.assertEqual(received_errors, [err])
341+
342+
def test_producer_reset_consumer_integration(self):
343+
"""Test producer reset clears consumer active errors."""
344+
with (
345+
self.assertLogs(level=logging.INFO),
346+
mock_rx_thread(
347+
self.consumer,
348+
lambda: self.producer.send(0x2001, 0x02, b'\x01\x02\x03\x04\x05'),
349+
),
350+
):
351+
self.consumer.wait(0x2001, timeout=TIMEOUT)
352+
self.assertEqual(len(self.consumer.active), 1)
353+
with (
354+
self.assertLogs(level=logging.INFO),
355+
mock_rx_thread(self.consumer, self.producer.reset),
356+
):
357+
self.assertIsNotNone(self.consumer.wait(timeout=TIMEOUT))
358+
self.assertEqual(len(self.consumer.active), 0)
359+
self.assertEqual(len(self.consumer.log), 2)
360+
223361

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

0 commit comments

Comments
 (0)