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

Question

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

一方面，我有一个点击命令行，它创建一个 grpclib channel 来点击 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 的异步上下文管理器，您可以在通过模拟调用的 Hook 中初始化上下文管理器，如下面的 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 ==============================