跳转到主要内容

pytest的插件,用于简化从测试或固定值调用ansible模块

项目描述

pytest-ansible

Build Status Version License Supported Python Versions

pytest-ansible 插件旨在提供 pytestAnsible 之间的无缝集成,让您能够高效地在 pytest 测试套件中运行和测试与 Ansible 相关的任务和场景。此插件通过提供三项不同的功能来增强测试工作流程:

  1. Ansible 集合的单元测试:此功能有助于使用 pytest 运行 Ansible 集合 的单元测试。它允许您单独验证 Ansible 模块角色 的行为,确保每个组件按预期工作。

  2. Molecule 场景集成:该插件帮助使用 pytest 运行 Molecule 场景。此集成简化了在不同环境中对 Ansible 角色和剧本的测试,使识别和修复不同配置中的问题变得更容易。

  3. Ansible 集成到 Pytest 测试:使用此功能,您可以在 pytest 测试中无缝使用 Ansible。这打开了与 Ansible 组件交互以及执行诸如资源部署、配置测试等任务的可能性,同时利用 pytest 的强大功能和灵活性。

支持的 Ansible

Pytest Ansible 只支持处于 活跃上游支持pythonansible-core 版本,目前这相当于:

  • Python 3.10 或更高版本
  • Ansible-core 2.14 或更高版本

安装

使用 pip 安装此插件

pip install pytest-ansible

入门指南

Ansible 集合的单元测试

pytest-ansible-units 插件允许使用 pytest 仅运行 Ansible 集合的单元测试。它提供了一种针对单个 Ansible 模块进行测试的专注方法。使用此插件,您可以编写和执行针对 Ansible 模块的特定单元测试,确保模块代码的准确性和可靠性。这对于验证模块行为的正确性非常有用。

要使用 pytest-ansible-units,请按照以下步骤操作:

  1. 使用 pip 安装插件
pip install pytest-ansible
  1. 确保您已安装 Python 3.10 或更高版本、ansible-core 和 pyyaml。

  2. 根据您首选的目录结构,您可以将集合克隆到适当的路径。

    • 集合树方法:首选方法是克隆正在开发的集合到其正确的集合树路径。这消除了对任何符号链接的需要,并且可以克隆到相同的树结构中的其他正在开发的集合。

      git clone <repo> collections/ansible_collections/<namespace>/<name>
      

      注意:在集合目录的根目录中运行 pytest,紧邻集合的 galaxy.yml 文件。

    • 浅树方法:

      git clone <repo>
      

      注意

      • 在集合目录的根目录中运行 pytest,紧邻集合的 galaxy.yml 文件。
      • 将在存储库目录中创建一个集合目录,并将集合内容链接到其中。
  3. 使用 pytest 执行单元测试: pytest tests

帮助

以下内容可以添加到集合的 pyproject.toml 文件中,以限制警告并设置集合测试的默认路径

[tool.pytest.ini_options]
testpaths = [
    "tests",
]
filterwarnings = [
    'ignore:AnsibleCollectionFinder has already been configured',
]

galaxy.yml 文件中获取的信息用于构建 collections 目录结构和链接内容。galaxy.yml 文件应反映正确的集合命名空间和名称。

检测问题而不运行测试的一种方法是在以下命令中运行

pytest --collect-only

可能看到以下错误

E   ModuleNotFoundError: No module named 'ansible_collections'
  • 检查 galaxy.yml 文件以获取准确的命名空间和名称
  • 确保从集合的根目录运行 pytest,紧邻 galaxy.yml
HINT: remove __pycache__ / .pyc files and/or use a unique basename for your test file modules

分子场景集成

此功能有助于使用 pytest 运行 Molecule scenarios。它使 pytest 能够发现代码库内所有 molecule.yml 文件,并将它们作为 pytest 测试运行。它允许您将 Molecule 场景作为 pytest 测试套件的一部分,从而让您能够彻底测试您的 Ansible 角色和剧本在不同场景和环境中的表现。

使用 pytest 运行分子场景

如果已安装 molecule,则可以使用两种不同的方法测试分子场景。

推荐

向 ansible 集合的 tests/integration 目录中添加 test_integration.py 文件

"""Tests for molecule scenarios."""
from __future__ import absolute_import, division, print_function

from pytest_ansible.molecule import MoleculeScenario


def test_integration(molecule_scenario: MoleculeScenario) -> None:
    """Run molecule for each scenario.

    :param molecule_scenario: The molecule scenario object
    """
    proc = molecule_scenario.test()
    assert proc.returncode == 0

molecule_scenario 修复程序提供了在集合的 extensions/molecule 目录以及集合内的其他目录中发现的参数化分子场景。

将为每个场景运行 molecule test -s <scenario>,并从 test() 调用返回完成的子进程。

旧版

使用 --molecule 命令行参数运行 molecule,以将当前工作目录中找到的每个 molecule 目录注入。每个场景将作为 pytest 可用测试的外部测试注入。由于这种方法的特点,molecule 场景不会以 Python 测试的形式表示,可能不会显示在 IDE 的 pytest 测试树中。

要使用 pytest 运行 Molecule 场景,请按照以下步骤操作

  1. 使用 pip 安装 pytest-ansible 插件
pip install pytest-ansible
  1. 执行 pytest 运行 Molecule 场景:pytest

pytest 测试的 Ansible 集成

ansible_moduleansible_adhoclocalhostansible_facts 修复程序被提供以帮助您将 Ansible 功能集成到您的 pytest 测试中。这些修复程序允许您与 Ansible 模块交互、在本地主机上运行命令、获取 Ansible 事实信息等。

测试中使用的修复程序和辅助函数

以下是可用修复程序的快速概述

  • ansible_module:允许您在测试函数中直接调用 Ansible 模块。
  • ansible_adhoc:提供一个函数来初始化一个 HostManager 对象,以与 Ansible 清单一起使用。
  • localhost:一个方便的修复程序,用于运行通常在本地计算机上运行的 Ansible 模块。
  • ansible_facts:返回一个 JSON 结构,表示关联的清单的系统事实。

用法

安装后,以下 pytest 命令行参数可用

pytest \
    [--inventory <path_to_inventory>] \
    [--extra-inventory <path_to_extra_inventory>] \
    [--host-pattern <host-pattern>] \
    [--connection <plugin>] \
    [--module-path <path_to_modules] \
    [--user <username>] \
    [--become] \
    [--become-user <username>] \
    [--become-method <method>] \
    [--ask-become-pass] \
    [--limit <limit>] \
    [--ansible-unit-inject-only] \
    [--molecule] \
    [--molecule-unavailable-driver] \
    [--skip-no-git-change] \
    [--check]

清单

使用 ansible 首先需要定义您的清单。这可以通过多种方式完成,但为了开始,我们将使用 ansible_adhoc 修复程序。

def test_my_inventory(ansible_adhoc):
    hosts = ansible_adhoc()

在上面的示例中,hosts 变量是 HostManager 类的实例,并描述了您的 ansible 清单。为了使其正常工作,您需要告诉 ansible 在哪里找到您的清单。清单可以是任何由 ansible 支持的内容,包括一个 INI 文件 或一个可执行脚本,该脚本返回 正确格式化的 JSON。例如,

pytest --inventory my_inventory.ini --host-pattern all

pytest --inventory path/to/my/script.py --host-pattern webservers

pytest --inventory one.example.com,two.example.com --host-pattern all

在上面的示例中,在运行时提供的清单将用于所有使用 ansible_adhoc 修复程序的测试。更现实的情况可能涉及使用不同的清单文件(或主机模式)与不同的测试一起使用。为了实现这一点,ansible_adhoc 修复程序允许您自定义清单参数。请继续阅读有关使用 ansible_adhoc 修复程序的更多详细信息。

额外清单

使用Ansible首先需要定义额外的清单。这个功能是在版本2.3.0中添加的,旨在允许用户使用两个不同的清单。这可以通过几种方式完成,但为了开始,我们将使用ansible_adhoc设置。

例如:

pytest --inventory my_inventory.ini --extra-inventory my_second_inventory.ini --host-pattern host_in_second_inventory

设置ansible_adhoc

ansible_adhoc设置返回一个用于初始化HostManager对象的函数。默认情况下,ansible_adhoc设置将使用提供给pytest命令行的参数,但也允许提供用于初始化清单的关键字参数。

以下示例演示了在运行时向pytest提供选项的基本用法。

def test_all_the_pings(ansible_adhoc):
    ansible_adhoc().all.ping()

以下示例演示了创建HostManager对象时可用的关键字参数。

def test_uptime(ansible_adhoc):
    # take down the database
    ansible_adhoc(inventory='db1.example.com,', user='ec2-user',
        become=True, become_user='root').all.command('reboot')

ansible_adhoc()函数返回的HostManager对象提供了多种调用Ansible模块的方法,针对清单中的某些或所有主机。以下展示了示例用法。

def test_host_manager(ansible_adhoc):
    hosts = ansible_adhoc()

    # __getitem__
    hosts['all'].ping()
    hosts['localhost'].ping()

    # __getattr__
    hosts.all.ping()
    hosts.localhost.ping()

    # Supports [ansible host patterns](https://docs.ansible.org.cn/ansible/latest/intro_patterns.html)
    hosts['webservers:!phoenix'].ping()  # all webservers that are not in phoenix
    hosts[0].ping()
    hosts[0:2].ping()

    assert 'one.example.com' in hosts

    assert hasattr(hosts, 'two.example.com')

    for a_host in hosts:
        a_host.ping()

设置localhost

localhost设置是一个便利的设置,它提供了一个用于在运行pytest的本地Ansible主机上的ModuleDispatcher实例。当使用通常在本地机器上运行的Ansible模块时,例如云模块(ec2、gce等...),这非常方便。

def test_do_something_cloudy(localhost, ansible_adhoc):
    """Deploy an ec2 instance using multiple fixtures."""
    params = dict(
        key_name='some_key',
        instance_type='t2.micro',
        image='ami-123456',
        wait=True,
        group='webservers',
        count=1,
        vpc_subnet_id='subnet-29e63245',
        assign_public_ip=True,
    )

    # Deploy an ec2 instance from localhost using the `ansible_adhoc` fixture
    ansible_adhoc(inventory='localhost,', connection='local').localhost.ec2(**params)

    # Deploy an ec2 instance from localhost using the `localhost` fixture
    localhost.ec2(**params)

设置ansible_module

ansible_module设置允许测试和设置调用Ansible模块。与ansible_adhoc设置不同,此设置仅使用在运行时提供给pytest的选项。

以下是一个非常基础的示例,演示了Ansible的ping模块

def test_ping(ansible_module):
    ansible_module.ping()

以下是一个更复杂的示例,演示了更新sshd配置并重新启动服务。

def test_sshd_config(ansible_module):

    # update sshd MaxSessions
    contacted = ansible_module.lineinfile(
        dest="/etc/ssh/sshd_config",
        regexp="^#?MaxSessions .*",
        line="MaxSessions 150")
    )

    # assert desired outcome
    for (host, result) in contacted.items():
        assert 'failed' not in result, result['msg']
        assert 'changed' in result

    # restart sshd
    contacted = ansible_module.service(
        name="sshd",
        state="restarted"
    )

    # assert successful restart
    for (host, result) in contacted.items():
        assert 'changed' in result and result['changed']
        assert result['name'] == 'sshd'

    # do other stuff ...

设置ansible_facts

ansible_facts设置返回一个JSON结构,表示关联清单的系统事实。示例事实数据可在Ansible文档中找到。

请注意,此设置提供是为了方便,并且可以很容易地使用ansible_module.setup()调用。

系统事实在决定是否跳过测试时可能很有用...

def test_something_with_amazon_ec2(ansible_facts):
    for facts in ansible_facts:
        if 'ec2.internal' != facts['ansible_domain']:
            pytest.skip("This test only applies to ec2 instances")

此外,由于事实只是Ansible模块,您还可以检查ec2_facts模块的内容以获得更高的粒度...

def test_terminate_us_east_1_instances(ansible_adhoc):

    for facts in ansible_adhoc().all.ec2_facts():
        if facts['ansible_ec2_placement_region'].startswith('us-east'):
            '''do some testing'''

使用pytest.mark.ansible参数化

也许--ansible-inventory=<inventory>包括许多系统,但您只想与子集交互。可以使用pytest.mark.ansible标记修改单个测试的pytest-ansible命令行参数。请注意,ansible_adhoc设置是交互Ansible清单的测试中的首选机制。

例如,要与本地系统交互,您将调整host_patternconnection参数。

@pytest.mark.ansible(host_pattern='local,', connection='local')
def test_copy_local(ansible_module):

    # create a file with random data
    contacted = ansible_module.copy(
        dest='/etc/motd',
        content='PyTest is amazing!',
        owner='root',
        group='root',
        mode='0644',
    )

    # assert only a single host was contacted
    assert len(contacted) == 1, \
        "Unexpected number of hosts contacted (%d != %d)" % \
        (1, len(contacted))

    assert 'local' in contacted

    # assert the copy module reported changes
    assert 'changed' in contacted['local']
    assert contacted['local']['changed']

请注意,pytest.mark.ansible提供的参数将应用于所有类方法。

@pytest.mark.ansible(host_pattern='local,', connection='local')
class Test_Local(object):
    def test_install(self, ansible_module):
        '''do some testing'''
    def test_template(self, ansible_module):
        '''do some testing'''
    def test_service(self, ansible_module):
        '''do some testing'''

检查结果

当使用ansible_adhoclocalhostansible_module设置时,返回的对象将是AdHocResult类的实例。可以按照以下方式检查AdHocResult类:

def test_adhoc_result(ansible_adhoc):
    contacted = ansible_adhoc(inventory=my_inventory).command("date")

    # As a dictionary
    for (host, result) in contacted.items():
        assert result.is_successful, "Failed on host %s" % host
    for result in contacted.values():
        assert result.is_successful
    for host in contacted.keys():
        assert host in ['localhost', 'one.example.com']

    assert contacted.localhost.is_successful

    # As a list
    assert len(contacted) > 0
    assert 'localhost' in contacted

    # As an iterator
    for result in contacted:
        assert result.is_successful

    # With __getattr__
    assert contacted.localhost.is_successful

    # Or __getitem__
    assert contacted['localhost'].is_successful

使用AdHocResult对象提供了方便地访问Ansible adhoc命令中涉及的不同主机的结果的方法。一旦找到特定的主机结果,就可以通过ModuleResult接口检查该特定主机上Ansible adhoc命令的结果。ModuleResult类代表Ansible模块为特定主机返回的字典。字典的内容取决于调用的模块。

ModuleResult接口提供了一些方便的属性来确定模块调用的成功与否。以下包括了一些示例。

def test_module_result(localhost):
    contacted = localhost.command("find /tmp")

    assert contacted.localhost.is_successful
    assert contacted.localhost.is_ok
    assert contacted.localhost.is_changed
    assert not contacted.localhost.is_failed

    contacted = localhost.shell("exit 1")
    assert contacted.localhost.is_failed
    assert not contacted.localhost.is_successful

由Ansible模块返回的JSON内容因模块而异。对于指导,请参考特定Ansible模块的文档和示例。

异常处理

如果ansible无法连接到任何清单,将引发异常。

@pytest.mark.ansible(inventory='unreachable.example.com,')
def test_shutdown(ansible_module):

    # attempt to ping a host that is down (or doesn't exist)
    pytest.raises(pytest_ansible.AnsibleHostUnreachable):
        ansible_module.ping()

有时,只有单个主机不可达,而其他主机已正确返回数据。以下演示了如何捕获异常并检查结果。

@pytest.mark.ansible(inventory='good:bad')
def test_inventory_unreachable(ansible_module):
    exc_info = pytest.raises(pytest_ansible.AnsibleHostUnreachable, ansible_module.ping)
    (contacted, dark) = exc_info.value.results

    # inspect the JSON result...
    for (host, result) in contacted.items():
        assert result['ping'] == 'pong'

    for (host, result) in dark.items():
        assert result['failed'] == True

通信

请参阅文档中的通信部分,了解如何联系我们。

您还可以在Ansible 通信指南中找到更多信息。

贡献

我们非常欢迎贡献。可以使用tox运行测试,请在提交拉取请求之前确保覆盖率至少保持不变。

行为准则

请参阅Ansible 社区行为准则

许可

MIT许可证下分发,"pytest-ansible"是免费和开源软件

问题

如果您遇到任何问题,请提交问题,并附带详细描述。

项目详情


下载文件

下载适合您平台的文件。如果您不确定选择哪个,请了解更多关于安装软件包的信息。

源代码分发

pytest_ansible-24.9.0.tar.gz (62.5 kB 查看哈希值)

上传时间 源代码

构建分发

pytest_ansible-24.9.0-py3-none-any.whl (30.0 kB 查看哈希值)

上传时间 Python 3

支持者

AWS AWS 云计算和安全赞助商 Datadog Datadog 监控 Fastly Fastly CDN Google Google 下载分析 Microsoft Microsoft PSF赞助商 Pingdom Pingdom 监控 Sentry Sentry 错误记录 StatusPage StatusPage 状态页面