Возникают проблемы с написанием модульного теста для команды Django mange.py, которая имеет асинхронный метод

Дана следующая рабочая manage.py команда:

class Command(BaseCommand):

def add_arguments(self, parser):
    parser.add_argument('path', nargs='+', type=str)

def handle(self, *args, **options):
    path = os.path.abspath(options['path'][0])
    asyncio.run(self._main(path))

async def _main(self, path):
    if os.path.isfile(path):
        async for i in async_generator(path):
            self.stdout.write(self.style.SUCCESS('data "%s"' % i))
    elif os.path.isdir(path):
        for file in glob.glob(f"{path}/*.txt"):
            async for i in async_generator(file):
                self.stdout.write(self.style.SUCCESS('data "%s"' % i))

Эта команда работает при запуске через терминал и успешно записывает данные в stdout, однако когда я пытаюсь проверить это в тесте, оказывается, что _main по какой-то причине не выполняется. Если я перемещу self.stdout.write в handle, мы получим data.

Вот тест:

def test_command(temp_file):
data = b"""foo
    bar
    hello
    world
    """
tp = temp_file(data)

out = StringIO()
call_command(
    foo.Command(), f"path=test.txt", stdout=out
)
assert "data" in out.getvalue()
tp.close()

Подозреваю, что проблема связана с тем, что call_command не очень хорошо играет с asyncio, но если бы это было так, я бы ожидал какую-то ошибку.

Кто-нибудь сталкивался с этим раньше?

Вернуться на верх