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

Question

I'm trying to test a click command that is async from pytest, but I am hitting the limits of my knowledge of asyncio (or approaching the problem with a wrong architecture)

On one side, I have a click command line, that creates a grpclib channel to hit a 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

Now I'm trying to test things using pytest and 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)

My issue is that the async_cmd around main expects to call asyncio.run. But by the time the test_main method is called, a loop is already running (launched by pytest).

What should I do?

Should I modify my wrapper to join an existing loop (and how so?).
Should I mock something somewhere?
Should I just change my code do have my main just responsible for parsing the arguments and calling another function?

Stephen Rauch · Accepted Answer · 2021-05-16 23:01:48Z

5

You are running your own event loop in the async_cmd decorator with this:

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

Therefore, it is not apparent that you need to use @pytest.mark.asyncio, I suggest trying your testing without it.

If you need an async context manager for a Mock, you can init the context manager in a hook called via the mock as shown below in test_hook().

Test Code (for the 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 Results

============================= 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 ==============================

edited May 16, 2021 at 23:01

answered May 16, 2021 at 19:18

Stephen Rauch♦

50.1k32 gold badges118 silver badges143 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

Wam Over a year ago

Indeed, I don't need it if I only call runner.invoke. My issue is that I need to mock my async with Channel() as channel, which I intend to replace with a async ChannelFor() as test_channel that needs to be initialized in my test. And I can't call that before runner.invoke, given I'm not in an async fn.

Stephen Rauch Over a year ago

You should be able to init the ChannelFor in a hook via the mock. I have expanded the example to show this.

Wam Over a year ago

Returning the ChannelFor via the hook works indeed, and makes my outer async with unneeded, allowing the test method to be a regular one, not an async one. Thanks for your amazing and thorough reply.

WilliamStam · Accepted Answer · 2023-06-19 11:19:34Z

this is late and im pretty sure this wont be applicable but here goes:

if you use your own async event loop then call the .main() on the function you use for your click handler

main.py

import anyio
from app.cli import cli
if __name__ == '__main__':
    anyio.run(cli)

and app.cli can be something like:

async def cli:
    @click.group
    @click.pass_context
    async def cmd(ctx):
        if ctx.invoked_subcommand is None:
            print("fish cakes")

    await cmd.main()

source: https://pypi.org/project/asyncclick/

# You can use your own event loop:
#    import anyio
#    async def main():
#        await hello.main()
#    if __name__ == '__main__':
#        anyio.run(main)
# This is useful for testing.

(it took me a while to figure out that hello here is the click callback function and that asyncclick adds a .main() onto that click function to "not add an event loop")

this all is pretty obvious i guess.. but it had me tripped up for a while

Collectives™ on Stack Overflow

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

2 Answers 2

Test Code (for the test code)

Test Results

3 Comments

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Test Code (for the test code)

Test Results

3 Comments

Comments

Your Answer

Sign up or log in

Post as a guest

Related