如何执行异步函数的单元测试？答案

【问题标题】：How to perform unit tests of asynchronous functions?如何执行异步函数的单元测试？
【发布时间】：2019-06-18 07:33:32
【问题描述】：

我正在使用 Bleak 发现并连接到最近的蓝牙低功耗 (BLE) 设备，并且我目前正在编写单元测试（使用 pytest）。

我是 Python 测试的新手，我不知道如何处理这些补丁/模拟以使其适用于 async 函数。

我不知道是否应该使用实际功能，或者对默认功能应用补丁以使测试在没有 BLE 加密狗的情况下可执行。

这里是一个代码示例（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())

我想知道是否应该重写函数以将run() 部分移到外面，然后模拟它。

【问题讨论】：

标签： python unit-testing bluetooth-lowenergy pytest python-asyncio

【解决方案1】：

外部函数list(op_sys) -> list 不是异步的，因为它调用了loop.run_until_complete。

这样就可以像任何同步 python 函数一样进行单元测试。

如果您想对内部函数run() -> list 等异步函数进行单元测试，请查看此处：https://pypi.org/project/asynctest/。

【讨论】：

那么，在单元测试逻辑中，我不必测试run函数？（感谢异步测试，我会读的）
您将一个函数嵌套到另一个函数中。这对于隔离非常有用，可以帮助提高可读性并减少重复。但是复杂的逻辑应该放在unittest可以访问的地方。这些有时是相互矛盾的模式。这个函数在我看来你应该模拟 bleak.discover 并测试外部（同步）函数）。
我使用了解决方案described here，并且我的测试正在运行。我会用我的测试代码发布答案。

【解决方案2】：

所以对于 Freek 的 help，我知道我想模拟 bleak.discover，我是这样做的：

我使用 Ivan 的 this anwser 找到了解决方案。

这是我的测试：

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"

这是测试结果：

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

=== 1 passed in 0.09 seconds ===

编辑：优雅代码的建议：

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])
        ....

【讨论】：