Модульное тестирование асинцио сопрограмм - PullRequest
Новая
       6

Модульное тестирование асинцио сопрограмм

0 голосов
/ 08 апреля 2020

Я бы хотел использовать nosetests для проверки сопрограмм в классе. Моей первой мыслью было назвать каждую сопрограмму с asyncio.run(). К сожалению, я начал получать исключения во время выполнения после первого теста. Это минимальный пример, демонстрирующий проблему: 

import asyncio

class MyClass:

    def __init__(self):
        self.event = asyncio.Event()

    async def hello(self):
        print("Hello from MyClass")

class TestMyClass:

    def setup(self):
        self.my_class = MyClass()

    def test1(self):
        asyncio.run(self.my_class.hello())

    def test2(self):
        asyncio.run(self.my_class.hello())

Когда этот скрипт запускается с nosetests, RuntimeError: There is no current event loop in thread 'MainThread'. повышается для test2. Оскорбительная строка - это создание Event в MyClass. Я не до конца понимаю проблему. Документация для asyncio.run() гласит, что функция «должна использоваться в качестве основной точки входа для асинхронных программ и в идеале должна вызываться только один раз». Для меня это звучит так, как будто в программе можно вызывать asyncio.run() несколько раз, хотя это и не рекомендуется.

В любом случае, поскольку ошибка указывает на несуществующее событие l oop, я решил управлять событием l oop вручную и предложил следующий обходной путь: 

import asyncio

class MyClass:

    def __init__(self):
        self.event = asyncio.Event()

    async def hello(self):
        print("Hello from MyClass")

class TestMyClass:

    def __init__(self):
        try:
            self.loop = asyncio.get_event_loop()
        except RuntimeError:
            self.loop = asyncio.new_event_loop()

    def setup(self):
        asyncio.set_event_loop(self.loop)
        self.my_class = MyClass()

    def test1(self):
        asyncio.run(self.my_class.hello())

    def test2(self):
        asyncio.run(self.my_class.hello())

try...except в инициализации требуется, когда несколько тестовых сценариев выполняются последовательно. Я использую Python 3.7.6.

Мой обходной путь не выглядит мне очень чистым, и мне интересно, есть ли лучший подход.

1 Ответ

0 голосов
/ 08 апреля 2020

Потенциальный обходной путь, поскольку в любом случае я использую asyncio низкоуровневые API, может быть следующим: 

import asyncio

class MyClass:

    def __init__(self):
        self.event = asyncio.Event()

    async def hello(self):
        print("Hello from MyClass")

class TestMyClass:

    def __init__(self):
        self.loop = asyncio.get_event_loop()

    def setup(self):
        self.my_class = MyClass()

    def test1(self):
        self.loop.run_until_complete(self.my_class.hello())

    def test2(self):
        self.loop.run_until_complete(self.my_class.hello())

Есть ли какие-либо возможные проблемы с этим подходом?

...