PAYATHON 2026

如何为 pyinfra 自定义 fact、operation 和部署代码编写自动化测试?

数据小陈

可以直接用 pytest 编写明确的测试用例。针对一个自定义 fact、几个 operation 和一份较长的部署文件,可以先测试 fact 的输出解析、operation 的状态判断和部署后的实际结果。等用例数量增加后,再考虑使用测试生成器。

为什么不必照搬 pyinfra 的测试组织方式?

pyinfra 项目需要覆盖大量内置 fact 和 operation,统一的用例格式和自动发现可以减少重复代码。对于自己编写的少量逻辑,直接写出输入和预期结果通常更容易维护。

原问题引用的 3.x 分支会变化。目前该文件已使用 pyinfra_testing.operations.make_operation_tests 生成测试,内容可能与提问时不同。可以参考它,但不必照搬为项目的测试模板。pyinfra 的 operation 测试入口

自定义 fact:先测试输出解析

fact 的 process() 接收命令输出的行列表,可以直接调用,不需要 SSH 连接。命令能否在目标系统执行,留给集成测试验证。Writing Facts

下面以一个假设的版本查询 fact 为例,命令和解析规则需要替换为你的实际实现:

# my_facts.py
from pyinfra.api import FactBase


class AppVersion(FactBase):
    command = "myapp --version"

    def process(self, output):
        if not output:
            return None
        return output[0].strip()

对应的测试:

# tests/test_facts.py
import pytest

from my_facts import AppVersion


@pytest.mark.parametrize(
    "output, expected",
    [
        (["1.2.3"], "1.2.3"),
        (["  1.2.3  "], "1.2.3"),
        ([], None),
    ],
)
def test_app_version_process(output, expected):
    assert AppVersion().process(output) == expected

实际用例应覆盖真实命令输出、空输出和可能出现的格式变化。如果解析失败时应抛出异常,也要明确测试这一行为。直接调用 process([]) 只能验证空列表的解析结果,不能证明框架会如何处理命令执行失败。

自定义 operation:测试状态与动作之间的对应关系

operation 的核心是读取 fact,比较当前状态和目标状态,然后生成命令。Writing Operations

建议为每个自定义 operation 列出以下场景:

当前状态目标状态预期行为
资源不存在资源应存在生成创建动作
资源已经符合要求保持该状态不生成修改动作
资源存在但配置不同更新配置生成必要的更新动作
输入无效或资源类型冲突无法执行按约定报告错误

如果状态比较或命令构造较复杂,可以将这部分提取为普通 Python 函数,直接断言返回结果。operation 负责获取 fact 并调用这些函数,再通过少量集成测试检查两者是否正确衔接。

如果选择模拟 fact 后检查 operation 生成的命令,需要根据所安装的 pyinfra 版本建立执行上下文。不要假设调用带装饰器的 operation 就会返回普通命令列表,也不要依赖未确认的内部属性。固定依赖版本后,再复用与该版本相符的测试辅助代码。

部署文件:测试调用参数和关键分支

较长的部署文件可以逐步拆成配置计算函数和部署入口。配置计算使用普通单元测试;部署入口可以模拟 operation,检查传入的参数是否正确。

下面是一个不读取 host 上下文的简化示例:

# deploy.py
from pyinfra.operations import files

files.file(
    path="/tmp/myapp-ready",
    present=True,
)

可以用 runpy 执行入口,并替换其调用的 operation:

# tests/test_deploy.py
import runpy
from pathlib import Path
from unittest.mock import Mock

from pyinfra.operations import files


DEPLOY = Path(__file__).resolve().parents[1] / "deploy.py"


def test_deploy_requests_marker_file(monkeypatch):
    file_operation = Mock()
    monkeypatch.setattr(files, "file", file_operation)

    runpy.run_path(str(DEPLOY))

    file_operation.assert_called_once_with(
        path="/tmp/myapp-ready",
        present=True,
    )

这个测试只验证部署代码发出的调用,不能证明文件确实创建成功。如果部署文件还读取 host.data 或调用 host.get_fact(),就需要提供相应的测试输入。把相关判断提取成显式接收参数的函数,通常比模拟整个运行环境更省事。

集成测试:验证实际结果和重复部署

保留少量集成测试,在可重建的容器或虚拟机中运行真实部署。涉及 systemd、重启或内核功能时,应选择能提供这些能力的环境。

一次完整检查可以按以下顺序进行:

  1. 创建干净的测试目标。
  2. 运行部署,检查命令是否成功结束。
  3. 独立检查文件内容、权限、软件包或服务状态。
  4. 对同一个目标再次运行部署,检查应当幂等的 operation 是否仍产生修改。

第二次运行必须保留第一次部署后的状态。每次都重建目标,只能验证首次安装。server.shell 等命令式 operation 可能每次都执行,应根据设计单独判断,不能要求所有操作都无变化。Using Operations

--dry 可以辅助检查计划,但无法验证实际写入、权限修改或服务启动是否成功,不能替代真实执行。使用 @local 时,修改会发生在运行命令的本机,应将这类测试放在专用的隔离环境中。Using Operations、@local Connector

对于目前的规模,可以先补齐 fact 解析测试、每个自定义 operation 的主要状态分支测试,以及一条完整部署的集成测试。示例用于说明组织方式,接入时需要按现有代码和安装版本调整。

备注:内容仅供参考。