如何在pytest异步测试中运行Typer（Tiangolo）命令？

我在manage.py中实现了一个createserviceuser的Typer异步命令，但用CliRunner测试时，result.stdout是空字符串，命令似乎没执行。相关代码如下：

manage.py #

import asyncio
from functools import wraps
from typing import Any

import typer
import uvicorn
from app.config import settings
from IPython import embed
from typing_extensions import ParamSpec


typer_app = typer.Typer()


class UsernameNotSetError(Exception):
    pass


class PasswordNotSetError(Exception):
    pass


class UserNotFound(Exception):
    pass


class ServiceUserRepository:

    def __init__(self, session: AsyncSession):
        self._session = session

    async def create_user(self, data):
        """创建服务用户。

        抛出异常：
            UsernameNotSetError
            PasswordNotSetError

        """

    async def check_user_created(self, username):
        """检查用户是否存在

        抛出异常：
            UserNotFound

        """


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


@typer_app.command()
@coro
async def createserviceuser(username: str, password: str, email: str = None):
    try:
        async with Session.begin() as session:
            resository = ServiceUserRepository(session)
            await resository.create_service_user(username, password, email)
    except UsernameNotSetError:
        print("Username not set")
        raise typer.exit
    except PasswordNotSetError:
        print("Password not set")
        raise typer.exit

    print("User created")


if __name__ == "__main__":
    typer_app()

测试代码 #

conftest.py #

def service_user_repository(session):
    return ServiceUserRepository(session)

test_commands.py #

from manage import typer_app

cli_runner = CliRunner()


class TestCreatedServiceUserCommand:

    async def test_username_not_set(self, service_user_repository):
        empty_username = ""
        result = cli_runner.invoke(typer_app, ["createserviceuser", empty_username, ""])
        # 异步命令未调用时，result.stdout为空字符串
        assert "Username not set" in result.stdout

        with pytest.raises(UserNotFound):
            await service_user_repository.check_user_exists(empty_username)

问题：调用CliRunner.invoke后，result.stdout为空，Typer命令未执行，该如何修复？

问题由多个代码错误和逻辑疏漏导致，逐一解决如下：

1. 修复Typer退出函数的调用方式 #

typer.exit是一个函数，必须调用它（即typer.exit()）才能触发命令退出。原代码中未加括号，会导致流程异常终止，无法输出预期内容。修改异常处理块：

except UsernameNotSetError:
    print("Username not set")
    raise typer.exit()  # 补全括号调用函数
except PasswordNotSetError:
    print("Password not set")
    raise typer.exit()

2. 修正方法名和变量名的拼写错误 #

• 仓库类中定义的方法是create_user，但命令中调用的是create_service_user，属于方法名不匹配，会直接抛出AttributeError，需统一方法名（要么修改仓库方法为create_service_user，要么命令中调用create_user）。
• 变量名resository拼写错误，应改为repository，避免后续逻辑报错。

3. 补全仓库方法的异常抛出逻辑 #

当前create_service_user（或create_user）是空实现，需要添加检查逻辑来抛出预期异常：

async def create_service_user(self, username: str, password: str, email: str = None):
    if not username.strip():
        raise UsernameNotSetError()
    if not password.strip():
        raise PasswordNotSetError()
    # 后续用户创建的数据库逻辑...

4. 修正测试中的方法名 #

测试代码中调用的check_user_exists，在仓库类中定义的是check_user_created，需统一名称：

with pytest.raises(UserNotFound):
    await service_user_repository.check_user_created(empty_username)

5. 确保异步测试环境正常运行 #

要让pytest支持异步测试，需安装pytest-asyncio插件，并修改conftest.py中的fixture为异步：

import pytest
from pytest_asyncio import fixture

@pytest.fixture
async def service_user_repository(session):
    return ServiceUserRepository(session)

6. 完善Session的导入与初始化 #

确保AsyncSession已正确导入并初始化，比如从SQLAlchemy导入：

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from app.config import settings

engine = create_async_engine(settings.DATABASE_URL)

async def get_session():
    async with AsyncSession(engine) as session:
        yield session

然后在命令中使用正确的Session获取方式，避免Session未定义的错误。

完成以上修改后，CliRunner.invoke就能正常执行命令，result.stdout会输出预期的"Username not set"，测试也能顺利通过。

内容的提问来源于stack exchange，提问作者Альберт Александров