Блог

Я пытаюсь протестировать click асинхронную команду из pytest, но я достигаю пределов своих знаний об asyncio (или приближаюсь к проблеме с неправильной архитектурой).

С одной стороны, у меня есть командная строка click, которая создает grpclib канал для доступа к api grpc.

 import asyncio
from grpclib import Channel
from functools import wraps

def async_cmd(func):
  @wraps(func)
  def wrapper(*args, **kwargs):
    return asyncio.run(func(*args, **kwargs))
  return wrapper

@click.command
@async_cmd
async def main():
  async with Channel('127.0.0.1', 1234) as channel:
    blah = await something(channel)
    do_stuff_with(blah)
  return 0
 

Теперь я пытаюсь протестировать вещи с помощью pytest и pytest-asyncio:

 from click.testing import CliRunner
from cli import main
from grpclib.testing import ChannelFor
import pytest

@pytest.mark.asyncio
async def test_main()
  async with ChannelFor([Service()]) as test_channel:
    # Plan is to eventually mock grpclib.Channel with test_channel here.
    runner = CliRunner()
    runner.invoke(main)
 

Моя проблема в том, что async_cmd вокруг main ожидает звонка asyncio.run .
Но к моменту test_main вызова метода цикл уже запущен (запущен pytest).

Что мне делать?

• Должен ли я изменить свою оболочку, чтобы присоединиться к существующему циклу (и как это сделать?).
• Должен ли я где-нибудь над чем-то посмеяться?
• Должен ли я просто изменить свой код, чтобы я main просто отвечал за анализ аргументов и вызов другой функции?

Вы запускаете свой собственный цикл событий в async_cmd декораторе с помощью этого:

 asyncio.run(func(*args, **kwargs))
 

Поэтому не очевидно, что вам нужно использовать @pytest.mark.asyncio , я предлагаю попробовать ваше тестирование без него.

Если вам нужен асинхронный контекстный менеджер для макета, вы можете инициализировать контекстный менеджер в крючке, вызываемом через макет, как показано ниже в test_hook() .

Тестовый код (для тестового кода)

 import asyncio
import click
import functools as ft
import pytest
import time
from unittest import mock
from click.testing import CliRunner


class AsyncContext():

    def __init__(self, delay):
        self.delay = delay

    async def __aenter__(self):
        await asyncio.sleep(self.delay)
        return self.delay

    async def __aexit__(self, exc_type, exc, tb):
        await asyncio.sleep(self.delay)

TestAsyncContext = AsyncContext

def async_cmd(func):
    @ft.wraps(func)
    def wrapper(*args, **kwargs):
        return asyncio.run(func(*args, **kwargs))
    return wrapper


@click.command()
@async_cmd
async def cli():
    async with TestAsyncContext(0.5) as delay:
        await asyncio.sleep(delay)
    print('hello')


@pytest.mark.parametrize('use_mock, min_time, max_time',
                         ((True, 2.5, 3.5), (False, 1.0, 2.0)))
def test_async_cli(use_mock, min_time, max_time):
    def test_hook(delay):
        return AsyncContext(delay   0.5)

    runner = CliRunner()
    start = time.time()
    if use_mock:
        with mock.patch('test_code.TestAsyncContext', test_hook):
            result = runner.invoke(cli)
    else:
        result = runner.invoke(cli)
    stop = time.time()
    assert result.exit_code == 0
    assert result.stdout == 'hellon'
    assert min_time < stop - start < max_time
 

Результаты испытаний

 ============================= test session starts =============================
collecting ... collected 2 items

test_code.py::test_async_cli[True-2.5-3.5] 
test_code.py::test_async_cli[False-1.0-2.0] 

============================== 2 passed in 4.57s ==============================