How to perform unit tests of asynchronous functions?

Question

I'm using Bleak to discover & connect to the nearest Bluetooth Low Energy (BLE) device, and I'm currently writing unit tests (using pytest).

I am new to Python tests and I don't know what to do with these patch/mock to make it work on async functions.

I do not know if I should use the actual function, or apply patches to the default functions to make the test executable without the BLE dongle.

Here is a sample of code (improvement of discover.py) :

def list(op_sys: str) -> list:
    """list BLE devices

    Returns:
        list: status & list or error message
    """
    import asyncio, platform
    from bleak import discover

    async def run() -> list:
        """discover BLE devices

        Returns:
            list: status & list or error message
        """
        BLElist = []
        try:
            devices = await discover()
            for d in devices:
                print("'%s'" % d.name) # list devices
                BLElist.append(d.name)
            return 'success', BLElist
        except:
            return 'error', 'You don\'t have any BLE dongle.'

    # linux = 3.6, windows = 3.7, need a new loop to work
    if op_sys == "Windows":
        asyncio.set_event_loop(asyncio.new_event_loop())

    loop = asyncio.get_event_loop()
    return loop.run_until_complete(run())

I'm wondering if I should rewrite the function to move the run() part outside, and mock it.

score 1 · Accepted Answer · answered Jun 18 '19 at 07:44

1

The outer function list(op_sys) -> list is not async because it does a call to loop.run_until_complete.

So that one can be unit tested like any synchronous python function.

If you want to unit test async functions like the inner function run() -> list, take a look over here: https://pypi.org/project/asynctest/.

answered Jun 18 '19 at 07:44

Freek Wiekmeijer

4,556
30
37

So, in the unit test logic, I do not have to test the `run` function? (thanks for asynctest, i will read that) – sodimel Jun 18 '19 at 07:52
1

You're nesting a function into another function. This is great for isolation and can help readability and reduce repetition. But complex logic should be in a place where unittest can access it. These are sometimes contradictory patterns. This function looks to me like you should mock `bleak.discover` and test the outer (synchronous) function). – Freek Wiekmeijer Jun 18 '19 at 08:10
I used solution [described here](https://stackoverflow.com/a/50031903/6813732), and my tests are working. I will post an answer with the code of my test. – sodimel Jun 18 '19 at 08:51

sodimel · Answer 2 · 2019-06-19T07:18:27.480

So with the help of Freek, I knew that I wanted to mock bleak.discover, and here is how I did it:

I found a solution using this anwser of Ivan.

Here is my test:

import os, asyncio
from unittest.mock import Mock
from app.functions.ble import ble

class BLE:
    def __init__(self, name):
        self.name = name

# code of Ivan, thank you Ivan!
def async_return(result):
    f = asyncio.Future()
    f.set_result(result)
    return f

def test_list(monkeypatch):

    mock_discover = Mock(return_value=async_return([BLE("T-000001"), BLE("T-000002"), BLE("T-000003")]))
    monkeypatch.setattr('bleak.discover', mock_discover)
    list_BLE = ble.list("Linux")

    mock_discover.assert_called_once()
    assert list_BLE[0] == 'success'
    assert list_BLE[1][0] == "T-000001"

And here is the test result:

tests/test_ble.py::test_list 'T-000001'
'T-000002'
'T-000003'
PASSED

=== 1 passed in 0.09 seconds ===

Edit: suggestion for elegant code:

from unittest import TestCase
from unittest.mock import patch
import os, asyncio

from app.functions.ble import ble


class DeviceDiscoveryTest(TestCase):

    @staticmethod
    def __async_return(result):
        f = asyncio.Future()
        f.set_result(result)
        return f

   @classmethod
   def mocked_discover(cls):
        return cls.__async_return([BLE("T-000001"), BLE("T-000002"), BLE("T-000003")])

    @patch('bleak.discocver', new=DeviceDiscoveryTest.mocked_discover)
    def test_discover_devices(self):
        list_BLE = ble.list("Linux")
        self.assertEquals('success', list_BLE[0])
        ....

How to perform unit tests of asynchronous functions?

2 Answers2