meshcore-dev/meshcore_py
1
0
Fork 0
mirror of https://github.com/meshcore-dev/meshcore_py.git synced 2026-04-20 22:13:49 +00:00
Code Issues Projects Releases Packages Wiki Activity

Merge 75c4a58841 into fbf84cbdac

Browse source
This commit is contained in:
mwolter805 2026-04-18 05:10:42 -07:00 • committed by GitHub
commit af47e4d848
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 11 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

11
tests/unit/test_commands.py
View file

@ -28,6 +28,10 @@ def mock_dispatcher():
sub.unsubscribe = MagicMock()
dispatcher._last_subscribe_handler = handler
dispatcher._last_subscribe_event_type = event_type
# Immediately resolve the future so send() doesn't block
asyncio.get_event_loop().call_soon(
handler, Event(event_type, {})
)
return sub
dispatcher.subscribe = MagicMock(side_effect=fake_subscribe)
@ -80,6 +84,13 @@ async def test_send_with_event(command_handler, mock_connection, mock_dispatcher
async def test_send_timeout(command_handler, mock_connection, mock_dispatcher):
# Override to NOT resolve events, so we can test the timeout path
def non_resolving_subscribe(event_type, handler, attribute_filters=None):
sub = MagicMock(spec=Subscription)
sub.unsubscribe = MagicMock()
return sub
mock_dispatcher.subscribe = MagicMock(side_effect=non_resolving_subscribe)
result = await command_handler.send(b"test_command", [EventType.OK], timeout=0.1)
assert result.type == EventType.ERROR
assert result.payload == {"reason": "no_event_received"}