Как выполнять модульные тесты асинхронных функций?

Я использую Bleak для обнаружения и подключения к ближайшему устройству Bluetooth с низким энергопотреблением (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 python-asyncio pytest bluetooth-lowenergy
person sodimel    schedule 18.06.2019    source источник

Ответы (2)


arrow_upward
2
arrow_downward

Внешняя функция list(op_sys) -> list не является асинхронной, потому что она вызывает loop.run_until_complete.

Так что можно протестировать модуль, как любую синхронную функцию Python.

Если вы хотите протестировать асинхронные функции, такие как внутренняя функция run() -> list, посмотрите здесь: https://pypi.org/project/asynctest/.

person Freek Wiekmeijer    schedule 18.06.2019
comment
Итак, в логике модульного теста мне не нужно тестировать функцию run? (спасибо за asynctest, я прочитаю это) - person sodimel; 18.06.2019
comment
Вы вкладываете функцию в другую функцию. Это отлично подходит для изоляции и может улучшить читаемость и уменьшить количество повторений. Но сложная логика должна быть там, где к ней может получить доступ юниттест. Иногда это противоречивые модели. Эта функция выглядит так, как будто вы должны издеваться над bleak.discover и тестировать внешнюю (синхронную) функцию). - person Freek Wiekmeijer; 18.06.2019
comment
Я использовал решение, описанное здесь, и мои тесты работают. Я опубликую ответ с кодом моего теста. - person sodimel; 18.06.2019

arrow_upward
0
arrow_downward

Итак, с помощью Фрика я знал, что Я хотел издеваться над bleak.discover, и вот как я это сделал:

Я нашел решение, используя этот ответ Ивана.

Вот мой тест:

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])
        ....
person Community    schedule 18.06.2019