Skip to content

Add unit tests for emcy.py (coverage improved from 73% → 99%) #604

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from

Conversation

yuvraj-kolkar17
Copy link

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:

Fixes #514


image

Does this PR introduce a user-facing change?

NONE

@yuvraj-kolkar17 yuvraj-kolkar17 mentioned this pull request Jul 21, 2025
23 tasks
@yuvraj-kolkar17
Copy link
Author

Hi @acolomb
Just wanted to kindly follow up on this PR. Please let me know if there’s anything I need to improve or change. I'd love to contribute more to this project

@acolomb
Copy link
Member

acolomb commented Jul 27, 2025

Will do as time permits.

Copy link

codecov bot commented Jul 27, 2025

Codecov Report

✅ All modified and coverable lines are covered by tests.

📢 Thoughts on this report? Let us know!

@yuvraj-kolkar17
Copy link
Author

yuvraj-kolkar17 commented Jul 27, 2025

Will do as time permits.

Thanks for the update @acolomb ! I’ll wait for your review.
Does this project have an official Slack or community channel for contributors?

@acolomb
Copy link
Member

acolomb commented Jul 27, 2025

Does this project have an official Slack or community channel for contributors?

None that I know of. Mainly GitHub for discussing code.

@yuvraj-kolkar17
Copy link
Author

Does this project have an official Slack or community channel for contributors?

None that I know of. Mainly GitHub for discussing code.

Got it, thanks for clarifying!

Copy link
Member

@acolomb acolomb left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@@ -25,13 +26,11 @@ def check_error(self, err, code, reg, data, ts):
self.assertAlmostEqual(err.timestamp, ts)

def test_emcy_consumer_on_emcy(self):
# Make sure multiple callbacks receive the same information.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd rather see this converted to a docstring than removing.

acc1 = []
acc2 = []
self.emcy.add_callback(lambda err: acc1.append(err))
self.emcy.add_callback(lambda err: acc2.append(err))

# Dispatch an EMCY datagram.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is fine, no real added value in the comment.

@@ -94,24 +91,20 @@ def timer(func):
finally:
t.join(TIMEOUT)

# Check unfiltered wait, on timeout.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@@ -123,6 +116,110 @@ def push_reset():
t.start()
self.assertIsNone(self.emcy.wait(0x9000, TIMEOUT))

def test_emcy_consumer_initialization(self):
"""Test EmcyConsumer initialization state."""
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Now this is pretty much redundant with the function name. 😉

Suggested change
"""Test EmcyConsumer initialization state."""

self.assertEqual(consumer.log, [])
self.assertEqual(consumer.active, [])
self.assertEqual(consumer.callbacks, [])
self.assertIsInstance(consumer.emcy_received, threading.Condition)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +428 to +440
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)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Suggested change
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)

Comment on lines +441 to +457
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)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Suggested change
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)

Comment on lines +486 to +490
msg = self.rxbus.recv(TIMEOUT)
self.assertIsNotNone(msg)

self.consumer.on_emcy(msg.arbitration_id, msg.data, msg.timestamp)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not use the regular flow via network.subscribe()?

Comment on lines +507 to +511
msg = self.rxbus.recv(TIMEOUT)
self.assertIsNotNone(msg)

self.consumer.on_emcy(msg.arbitration_id, msg.data, msg.timestamp)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why not use the regular flow via network.subscribe()?


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

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please watch out for accidental whitespace changes and inconsistencies.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

Improve test coverage
2 participants