如何从异步 pytest 函数测试异步单击命令？

Question

我正在尝试测试一个与 pytest 异步的 click 命令，但我正在达到我对 asyncio 知识的限制（或使用错误的体系结构解决问题）

一方面，我有一个点击命令行，它创建了一个 grpclib 通道来点击 grpc api。

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)

我的问题是 main 周围的 async_cmd 期望调用 asyncio.run。但是在调用 test_main 方法时，循环已经运行（由 pytest 启动）。

我该怎么办？

我是否应该修改我的包装器以加入现有循环（以及如何做？）。
我应该在某个地方模拟一些东西吗？
我应该只更改我的代码让我的 main 只负责解析参数并调用另一个函数吗？

Answer 1

你是运行你自己在 async_cmd 装饰器中的事件循环：

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

因此，您并不明显需要使用 @pytest.mark.asyncio，我建议您在没有它的情况下进行测试。

如果您需要 Mock 的异步上下文管理器，您可以在通过 mock 调用的挂钩中初始化上下文管理器，如下面 test_hook().

中所示

Test Code（为测试代码）

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 == 'hello\n'
    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 ==============================

如何从异步 pytest 函数测试异步单击命令？

How can I test async click commands from an async pytest function?

python

command-line-interface

pytest

python-asyncio

python-click

Test Code（为测试代码）

测试结果