Add unit tests for emcy.py (coverage improved from 73% → 99%)#604
Add unit tests for emcy.py (coverage improved from 73% → 99%)#604yuvraj-kolkar17 wants to merge 1 commit intocanopen-python:masterfrom
Conversation
|
Hi @acolomb |
|
Will do as time permits. |
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
Thanks for the update @acolomb ! I’ll wait for your review. |
None that I know of. Mainly GitHub for discussing code. |
Got it, thanks for clarifying! |
acolomb
left a comment
There was a problem hiding this comment.
Sorry for taking so long with the review. I started several attempts, but never got through to the end until now.
Some of these tests seem unnecessary IMO, see inline comments. And please make sure to follow a clean coding / formatting style.
| self.assertAlmostEqual(err.timestamp, ts) | ||
|
|
||
| def test_emcy_consumer_on_emcy(self): | ||
| # Make sure multiple callbacks receive the same information. |
There was a problem hiding this comment.
I'd rather see this converted to a docstring than removing.
| self.emcy.add_callback(lambda err: acc1.append(err)) | ||
| self.emcy.add_callback(lambda err: acc2.append(err)) | ||
|
|
||
| # Dispatch an EMCY datagram. |
There was a problem hiding this comment.
This is fine, no real added value in the comment.
| finally: | ||
| t.join(TIMEOUT) | ||
|
|
||
| # Check unfiltered wait, on timeout. |
There was a problem hiding this comment.
I haven't check whack happens here, but it sounds like it may be harder to understand without the comment. So please try to leave other code untouched unless it's an obviously superfluous comment.
Same below a couple of times.
| self.assertIsNone(self.emcy.wait(0x9000, TIMEOUT)) | ||
|
|
||
| def test_emcy_consumer_initialization(self): | ||
| """Test EmcyConsumer initialization state.""" |
There was a problem hiding this comment.
Now this is pretty much redundant with the function name. 😉
| """Test EmcyConsumer initialization state.""" |
| self.assertEqual(consumer.log, []) | ||
| self.assertEqual(consumer.active, []) | ||
| self.assertEqual(consumer.callbacks, []) | ||
| self.assertIsInstance(consumer.emcy_received, threading.Condition) |
There was a problem hiding this comment.
I don't really get what important characteristics this function tests for. The emcy_received attribute being of a specific type seems to me like an implementation detail, not public, dependable API.
| def test_emcy_producer_network_assignment(self): | ||
| """Test EmcyProducer network assignment and usage.""" | ||
| producer = canopen.emcy.EmcyProducer(0x100) | ||
| initial_network = producer.network | ||
|
|
||
| producer.network = self.net | ||
| self.assertEqual(producer.network, self.net) | ||
|
|
||
| producer.send(0x1000) | ||
| msg = self.rxbus.recv(TIMEOUT) | ||
| self.assertIsNotNone(msg) | ||
| self.assertEqual(msg.arbitration_id, 0x100) | ||
|
|
There was a problem hiding this comment.
I think this test is pretty redundant. We are using send and receive functions in several other tests, and this basically just asserts that assigning an attribute results in the value being assigned.
| def test_emcy_producer_network_assignment(self): | |
| """Test EmcyProducer network assignment and usage.""" | |
| producer = canopen.emcy.EmcyProducer(0x100) | |
| initial_network = producer.network | |
| producer.network = self.net | |
| self.assertEqual(producer.network, self.net) | |
| producer.send(0x1000) | |
| msg = self.rxbus.recv(TIMEOUT) | |
| self.assertIsNotNone(msg) | |
| self.assertEqual(msg.arbitration_id, 0x100) |
| def test_emcy_producer_struct_packing(self): | ||
| """Test that the EMCY_STRUCT packing works correctly.""" | ||
| from canopen.emcy import EMCY_STRUCT | ||
|
|
||
| packed = EMCY_STRUCT.pack(0x1234, 0x56, b'\xAB\xCD\xEF\x12\x34') | ||
| expected = b'\x34\x12\x56\xAB\xCD\xEF\x12\x34' | ||
| self.assertEqual(packed, expected) | ||
|
|
||
| code, register, data = EMCY_STRUCT.unpack(expected) | ||
| self.assertEqual(code, 0x1234) | ||
| self.assertEqual(register, 0x56) | ||
| self.assertEqual(data, b'\xAB\xCD\xEF\x12\x34') | ||
|
|
||
| packed = EMCY_STRUCT.pack(0x1234, 0x56, b'\xAB') | ||
| expected = b'\x34\x12\x56\xAB\x00\x00\x00\x00' | ||
| self.assertEqual(packed, expected) | ||
|
|
There was a problem hiding this comment.
This one is really testing an internal implementation detail, namely the structure used for decoding. That is already covered by most other tests. It could be solved differently, then this test might fail, although the public API to be tested is still functional.
| def test_emcy_producer_struct_packing(self): | |
| """Test that the EMCY_STRUCT packing works correctly.""" | |
| from canopen.emcy import EMCY_STRUCT | |
| packed = EMCY_STRUCT.pack(0x1234, 0x56, b'\xAB\xCD\xEF\x12\x34') | |
| expected = b'\x34\x12\x56\xAB\xCD\xEF\x12\x34' | |
| self.assertEqual(packed, expected) | |
| code, register, data = EMCY_STRUCT.unpack(expected) | |
| self.assertEqual(code, 0x1234) | |
| self.assertEqual(register, 0x56) | |
| self.assertEqual(data, b'\xAB\xCD\xEF\x12\x34') | |
| packed = EMCY_STRUCT.pack(0x1234, 0x56, b'\xAB') | |
| expected = b'\x34\x12\x56\xAB\x00\x00\x00\x00' | |
| self.assertEqual(packed, expected) |
| msg = self.rxbus.recv(TIMEOUT) | ||
| self.assertIsNotNone(msg) | ||
|
|
||
| self.consumer.on_emcy(msg.arbitration_id, msg.data, msg.timestamp) | ||
|
|
There was a problem hiding this comment.
Why not use the regular flow via network.subscribe()?
| msg = self.rxbus.recv(TIMEOUT) | ||
| self.assertIsNotNone(msg) | ||
|
|
||
| self.consumer.on_emcy(msg.arbitration_id, msg.data, msg.timestamp) | ||
|
|
There was a problem hiding this comment.
Why not use the regular flow via network.subscribe()?
|
|
||
| if __name__ == "__main__": | ||
| unittest.main() | ||
|
No newline at end of file |
There was a problem hiding this comment.
Please watch out for accidental whitespace changes and inconsistencies.
|
@yuvraj-kolkar17 Are you still up to work on this improvement? |
@acolomb Sorry for the delay but I’m currently in the middle of my university exams, so I won’t be able to work on this right away. I’ll get back to it as soon as my exams are over. |
What type of PR is this?
/kind test
/area emcy.py
What this PR does / why we need it:
Adds a complete unit test suite for the emcy.py component under
canopen/emcy.py.This test ensures core EMCY functionalities work correctly and helps future development by validating key scenarios.
Which issue(s) this PR fixes:
Related to #514
Does this PR introduce a user-facing change?