Asyncio тестирует функцию, которая использует l oop .call_soon_threadsafe

Question

У меня есть следующая функция:

def send_command(self, cmd):
    self.loop.call_soon_threadsafe(
        functools.partial(
            self._transport.write, str(cmd).encode() + b"\n"
        )
    )

Тестируемая система (sut) - это класс, унаследованный от asyncio.Protocol, который отправляет некоторые команды элементу оборудования на сокете. Я должен использовать темы, так как это часть GUI под wx Python. Наконец, если я позвоню self._transport.write, код отлично работает на Linux, но вылетает на Windows ™.

При запуске теста:

@pytest.mark.asyncio
async def test_send_command(self):
    self.sut._transport = Mock()
    self.sut.send_command("ook eek")
    assert self.sut._transport.write.called is True

Я получаю ошибку подтверждения. self.sut._transport.write никогда не называется. Если я вызываю self._transport.write непосредственно в функции, код вылетает в Windows ™, но тест проходит просто отлично.

Что мне здесь не хватает?

Кто-нибудь? …

Конечно, это не такой крайний случай…

Answer 1

Работа вокруг ...

После чтения и экспериментирования с четными циклами , используя это:

import asyncio
import selectors

selector = selectors.SelectSelector()
loop = asyncio.SelectorEventLoop(selector)
asyncio.set_event_loop(loop)

мимо проходит проблему. Конечно, это означает использование субэффективного l oop на Windows. ☹ PHA!

Любой, у кого есть лучшее решение, может получить фальшивые целые rnet баллы.