可运行示例：CLI 应用程序

当应用程序从命令行被调用时（例如 python -m myapp，使用 argparse/click 的 CLI 工具）。

做法：使用 asyncio.create_subprocess_exec 调用 CLI 并捕获输出。

# pixie_qa/run_app.py
import asyncio
import sys

from pydantic import BaseModel
import pixie


class AppArgs(BaseModel):
    query: str


class AppRunnable(pixie.Runnable[AppArgs]):
    """通过子进程驱动 CLI 应用程序。"""

    @classmethod
    def create(cls) -> "AppRunnable":
        return cls()

    async def run(self, args: AppArgs) -> None:
        proc = await asyncio.create_subprocess_exec(
            sys.executable, "-m", "myapp", "--query", args.query,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)
        if proc.returncode != 0:
            raise RuntimeError(f"App 失败（退出码 {proc.returncode}）：{stderr.decode()}")

当 CLI 需要修补的依赖项时

如果 CLI 读取外部服务，请创建一个包装入口点，在运行真实 CLI 之前修补依赖项：

# pixie_qa/patched_app.py
"""在运行真实 CLI 之前修补外部依赖项的入口点。"""
import myapp.config as config
config.redis_url = "mock://localhost"

from myapp.main import main
main()

然后将 Runnable 指向该包装器：

async def run(self, args: AppArgs) -> None:
    proc = await asyncio.create_subprocess_exec(
        sys.executable, "-m", "pixie_qa.patched_app", "--query", args.query,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=120)

注意：对于 CLI 应用程序，wrap(purpose="input") 注入仅当应用程序在同一进程中运行时才有效。如果使用子进程，则可能需要通过环境变量或配置文件传递测试数据。

2.0 KiB Raw Permalink Blame History

可运行示例：CLI 应用程序

当 CLI 需要修补的依赖项时

2.0 KiB

Raw Permalink Blame History