Ansible Pytest#

Ansible Pytest – это расширение для системы тестирования pytest, обеспечивающее ее бесшовную интеграцию с Ansible.

Основные возможности#

Основные возможности Ansible Pytest:

  • Модульное тестирование коллекций.

    С помощью Ansible Pytest вы можете проверить поведение отдельных ролей или целых коллекций Ansible, чтобы убедиться в корректности работы каждого компонента.

  • Интеграция с Ansible Molecule.

    Ansible Pytest может использовать сценарии Molecule в связке с pytest. Это позволяет тестировать роли и сценарии Ansible в различных средах, упрощая выявление и устранение ошибок.

  • Интеграция Ansible с тестами Pytest.

    Ansible Pytest позволяет использовать возможности Ansible в наборах тестов pytest, например, вызывать предоставляемые Ansible модули для выполнения операций с тестовыми стендами, запускать специальные команды (ad-hoc) и тому подобное.

Установка#

Для установки Ansible Pytest выполните следующие действия:

  1. Подключите репозиторий Astra Automation.

    Инструкция по подключению репозитория
    1. В каталоге /etc/apt/sources.list.d/ создайте файл astra-automation.list со ссылкой на репозиторий Astra Automation:

      deb https://dl.astralinux.ru/aa/aa-debs-for-alse-1.8 <version> main
      

      Вместо <version> необходимо подставить версию устанавливаемой платформы, например, 1.2.

      Доступные версии продукта опубликованы в таблице История обновлений.

    2. Обновите список доступных пакетов:

      sudo apt update
      
  2. Выполните команду:

    sudo apt-get install pytest-ansible --yes
    

Установка при отсутствии доступа к интернету описана в документе Средства тестирования.

Применение#

Утилиту можно применять для выполнения следующих задач:

  • Проверка тестового окружения на соответствие требованиям, необходимым для корректного применения ролей и выполнения сценариев.

    1. Любым удобным способом создайте тестовое окружение.

    2. Создайте тесты, собирающие данные о созданном окружении и проверяющие соответствие фактических характеристик окружения ожидаемым.

    3. Запустите тесты и убедитесь, что окружение развернуто именно в той конфигурации, которая нужна.

  • Сравнение фактических изменений в тестовом окружении с ожидаемыми.

    1. Любым удобным способом создайте тестовое окружение.

    2. Запустите сценарии Ansible, чтобы внести изменения в тестовое окружение.

    3. Создайте тесты, проверяющие состояние тестового окружения после его изменения с помощью сценариев Ansible.

    4. Запустите тесты и убедитесь, что ваши playbook вносят все необходимые изменения.

  • Использование модулей Ansible в тестах.

    Ansible Pytest позволяет выполнять в тестах pytest следующие операции:

    • вызов модулей Ansible;

    • сбор фактов;

    • управление инвентарем.

Принцип работы#

Набор тестов (test suite) pytest это файл с кодом на языке Python. Имя файла с набором тестов должно удовлетворять шаблонам:

  • test_*.py;

  • *_test.py.

При запуске pytest ищет файлы с подходящими именами в текущем каталоге и всех его подкаталогах. Более подробно порядок поиска файлов описан в документации Pytest.

В самом простом случае тест – это функция, название которой начинается с префикса test_, например:

def test_connection():
    # ...

Также тесты могут быть методами класса. В этом случае необходимо выполнение следующих условий:

  • название класса должно начинаться с префикса Test;

  • название метода должно начинаться с префикса test_.

class TestConnectionClass:
    def test_connection(self):
        # ...

    def test_nginx_service(self):
        # ...

В коде теста должно выполняться сравнение фактического значения параметра с ожидаемым с помощью инструкции assert, например:

def test_hostname(hostname):
    assert hostname == 'localhost'

Если условие не выполняется, тест считается непройденным.

Инвентарь#

Инвентарь в Ansible Pytest – это инвентарь Ansible в формате INI или JSON, либо сценарий, возвращающий список узлов в формате JSON.

Инвентарь может быть задан следующими способами:

  • указание пути к файлу с инвентарным списком в опции --inventory при запуске pytest:

    pytest --inventory <inventory> --ansible-host-pattern <pattern>
    
  • использование вспомогательного кода ansible_adhoc.

    Пример настройки инвентаря с помощью этого вспомогательного кода см. ниже.

Вспомогательный код#

В pytest вспомогательным кодом (fixtures) называются функции, упрощающие работу с тестовым окружением. Код вспомогательных функций запускается автоматически перед выполнением тестов и может быть использован повторно. Это позволяет сократить объем кода тестов и облегчает его сопровождение.

Элементы вспомогательного кода Ansible Pytest и их описания приводятся в справочнике.

Декоратор pytest.mark.ansible#

Декоратор pytest.mark.ansible используется для фильтрации списка узлов, на которых должен быть выполнен тест. Это может быть полезно, если инвентарный список содержит большое количество узлов, но отдельные тесты нужно выполнить только на некоторых из них.

Интеграция со сценариями Molecule#

Ansible Pytest обнаруживает все файлы molecule.yml в коде проекта и запускает задания из них как тесты. Таким образом можно включать сценарии Molecule в набор тестов pytest и тестировать содержимое Ansible в различных сценариях и окружениях.

Интеграция с Molecule может быть выполнена следующими способами:

  • Использование объекта molecule_scenario.

    В каталоге выбранной коллекции создайте файл 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 предоставляет сценарии Molecule, обнаруженные в каталоге extensions/molecule и других каталогах коллекции. Для каждого обнаруженного сценария выполняется команда тестирования:

    molecule test -s <scenario>
    

    Здесь <scenario> – путь к файлу сценария Molecule.

    pytest запускает каждый сценарий в отдельном процессе с помощью метода test(). Если код завершения процесса равен 0, тест считается пройденным успешно. Коды завершения, отличные от 0, считаются признаком ошибки.

  • Включение тестов pytest в блок verify сценария Molecule.

    В этом случае тесты pytest проверяют корректность развертывания стендов, сравнивая их фактические параметры с ожидаемыми. Это делает процесс CI/CD более надежным.

Примеры#

Для изучения возможностей Ansible Pytest создайте каталог, в котором будут размещаться файлы тестов. Команды запуска тестов выполняйте внутри этого каталога.

Выполнение команды ping на всех узлах#

Этот тест выполняет специальную команду ping на всех узлах, указанных в инвентаре:

test_all_ping.py#
def test_all_pings(ansible_adhoc):
    # Call ping() command for all nodes in inventory
    result = ansible_adhoc().all.ping()

    for node, outcome in result.items():
        if outcome.is_successful:
            print(f"Node {node} ping successfull: {outcome['ping']}")
        else:
            print(f"Node {node} ping failed: {outcome}")

Результаты выполнения команды сохраняются в переменной result. Метод items возвращает два списка: названия узлов и результаты выполнения команды. В цикле выполняется обход списков. Если поле is_successfull содержит значение True, тест считается пройденным успешно и в терминал выводится соответствующее сообщение. В противном случае выводится сообщение об ошибке.

Время работы системы с момента последней загрузки#

Команда uptime возвращает время, прошедшее с момента загрузки операционной системы. Приведенный ниже тест проверяет возможность выполнения команды uptime на узлах, указанных в файле inventory-special.ini.

При этом в коде теста настройки исходного инвентаря заменяются на указанные:

  • подключение к узлам выполняется от имени пользователя administrator;

  • используется повышение привилегий;

  • для выполнения команд с повышенными привилегиями используется учетная запись пользователя root.

test_uptime.py#
def test_uptime_with_custom_inventory(ansible_adhoc):
    """Run uptime command with custom inventory."""
    nodes = ansible_adhoc(
        inventory="inventory-special.ini",
        user="administrator",
        become=True,
        become_user="root",
    )

    # Print current settings
    print("HostManager settings:")
    print(nodes.options)

    # Run adhoc command `uptime`
    result = nodes.all.command("uptime")

    print("\nUptime:")
    for node, outcome in result.items():
        if outcome.is_successful:
            # Print uptime
            print(f"Node {node} uptime successful:\n{outcome['stdout']}")
        else:
            # Print error message
            print(f"Node {node} uptime failed:\n{outcome}")

        assert outcome.is_successful, f"Uptime command failed on node {node}"

Переменная result присваивает результаты выполнения команды uptime.

Метод items возвращает два списка: названия узлов и результаты выполнения команды. В цикле выполняется обход списков. Если поле outcome.is_successful содержит значение True, тест считается пройденным успешно, и в терминал выводится количество времени, прошедшего с момента загрузки ОС тестового узла. В противном случае выводится сообщение об ошибке.

Указание инвентарного списка в аргументах CLI для этого теста не требуется:

pytest --host-pattern all

Различные способы обращения к узлам#

Этот тест демонстрирует различные способы обращения к узлам:

test_host_manager.py#
def test_host_manager(ansible_adhoc):
    """Using HostManager for localhost and remote nodes."""

    nodes = ansible_adhoc()

    # Run ping command on all nodes
    nodes["all"].ping()

    # Run ping command only on localhost
    nodes["localhost"].ping()

    # Iterating nodes:
    for node in nodes.keys():
        result = nodes[node].ping()
        assert result[node].is_successful, f"Ping failed on node {node}"

        # Print successful ping
        print(f"Node {node} ping successful with result: {result[node]}")

Переменная nodes получает результаты выполнения функции ansible_adhoc(), подготавливающей окружение к использованию специальных команд.

Путем обращения к ключу all тест выполняет команду ping на всех узлах, указанных в инвентаре. Затем, путем обращения к ключу localhost`, тест выполняет команду ``ping только на локальном узле.

В цикле выполняется проход по списку узлов. Тест запускает команду ping для каждого очередного узла, а результат ее выполнения сохраняет в переменную result.

Тест проверяет значение поля is_successful. Если оно равно True, тест считается выполненным успешно.

Использование вспомогательного кода localhost#

Этот тест проверяет выполнение команды ping на локальном узле:

test_local_ping.py#
def test_local_ping(localhost):
    """Test ping() command on localhost."""

    result = localhost.ping()

    for node, outcome in result.items():
        assert outcome.get("ping") == "pong", f"Ping for node {node} is failed"

        print(f"Node {node} ping successful with result: {outcome}")

В отличие от предыдущих тестов, для получения доступа к локальному узлу вместо вспомогательного кода ansible_adhoc используется вспомогательный код localhost. Он также позволяет выполнять специальные команды, но только на локальном узле.

Указание инвентарного списка в аргументах CLI в этом случае не требуется:

pytest --ansible-host-pattern all

Использование модулей Ansible#

Этот тест демонстрирует использования модуля ansible.builtin.ping с помощью вспомогательного кода ansible_module.

test_ansible_module.py#
def test_ansible_module(ansible_module):
    """Call ping() command from Ansible module."""
    result = ansible_module.ping()

    for node, outcome in result.items():
        assert outcome.get("ping") == "pong", f"Ping on {node} is failed"
        print(f"Ping successful on node {node} with result: {outcome}")

Переменная result получает результаты выполнения команды ping на всех узлах. Метод items возвращает два списка: названия узлов и результаты выполнения команды. В цикле выполняется обход списков. Если результат выполнения команды ping равен pong, тест считается пройденным успешно. В противном случае выводится сообщение об ошибке.

Обработка фактов Ansible#

Этот тест собирает и выводит в терминал следующие сведения об узлах, на которых выполняется тестирование:

Факт Ansible

Описание

ansible_os_family

Семейство ОС

ansible_distribution

Дистрибутив ОС

ansible_distribution_version

Версия дистрибутива ОС

ansible_hostname

Доменное имя узла

ansible_architecture

Архитектура

ansible_processor_count

Количество CPU

ansible_memtotal_mb

Объем установленной оперативной памяти, МБ

ansible_default_ipv4

Основной адрес IPv4

ansible_default_ipv6

Основной адрес IPv6

ansible_uptime_seconds

Время в секундах с момента загрузки ОС

test_print_top_10_facts.py#
def test_print_top_10_facts(ansible_facts):
    """Show 10 Ansible facts."""

    important_facts = [
        "ansible_os_family",
        "ansible_distribution",
        "ansible_distribution_version",
        "ansible_hostname",
        "ansible_architecture",
        "ansible_processor_count",
        "ansible_memtotal_mb",
        "ansible_default_ipv4",
        "ansible_default_ipv6",
        "ansible_uptime_seconds",
    ]

    for node, facts in ansible_facts.items():
        print(f"\nImportant facts for node {node}:")

        for fact in important_facts:
            value = facts["ansible_facts"].get(fact, "Unknown")
            print(f"{fact}: {value}")

Метод items возвращает два списка: названия узлов и факты Ansible о них. В цикле выполняется обход этих списков. Из списка фактов извлекаются только факты, перечисленные в списке important_facts. Переменной value присваивается значение факта, если оно известно, и Unknown в противном случае.

Использование декоратора pytest.mark.ansible#

Этот тест демонстрирует применение декоратора pytest.mark.ansible для замены списка узлов из инвентарного списка inventory.ini на localhost:

test_pytest_mark_ansible.py#
import pytest


@pytest.mark.ansible(host_pattern="localhost", connection="local")
def test_local_ping(ansible_module):
    """Replace nodes from inventory to localhost."""

    result = ansible_module.ping()  # Only on localhost

    assert "localhost" in result, "Ping command on localhost failed"
    assert result["localhost"]["ping"] == "pong", "Node localhost returned not a pong"

Значения, указанные в опции CLI --ansible-host-pattern, игнорируются. Вместо этого фильтру узлов присваивается значение localhost, а типу подключения connection значение local.

Для запуска теста используйте команду:

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

Проверка результатов выполнения команд#

Этот тест демонстрирует использование методов класса AdHocResult для проверки результатов выполнения команд:

test_adhoc_result.py#
def test_adhoc_result(ansible_adhoc):
    """Run command date() and check results."""

    contacted = ansible_adhoc().all.command("date")

    # Walking over all nodes:
    for node, result in contacted.items():
        assert result.is_successful, f"Command is failed on node {node}"

    # Walking over result values
    for result in contacted.values():
        assert (
            result.is_successful
        ), f"Command is failed on one of the nodes with result {result}"

    # Check node name is 'localhost' or 'vm'
    for node in contacted.keys():
        assert node in ["localhost", "vm"], f"Unexpected {node} in results"

    # Check results on localhost only
    assert contacted.localhost.is_successful, "Command is failed on localhost"

    # Check nodes list length
    assert len(contacted) > 0, "Result list is empty"

    # Check localhost in result list
    assert "localhost" in contacted, "Host 'localhost' not found in results"

    # Check successful result on localhost with __getattr__
    assert (
        contacted.localhost.is_successful
    ), "Command is failed on localhost over __getattr__"

    # Check successful result on localhost with __getitem__
    assert contacted[
        "localhost"
    ].is_successful, "Command is failed on localhost over __getitem__"

Проверка параметров окружения#

Этот тест проверяет соответствие фактических параметров узлов ожидаемым:

test_adhoc_result.py#
def test_environment(ansible_facts):
    """Validation of environment characteristic."""

    # Expected values
    expected_cores = 2
    expected_memory_mb = 1014 * 8  # 8 GB
    expected_disk_size_gb = 20

    # Check nodes parameters
    for node, facts in ansible_facts.items():
        # Check CPU cores count
        actual_cores = facts["ansible_facts"].get("ansible_processor_cores", 0)
        assert (
            actual_cores == expected_cores
        ), f"Expected {expected_cores} cores, got {actual_cores} for node {node}"

        # Check RAM size
        actual_memory_mb = facts["ansible_facts"].get("ansible_memtotal_mb", 0)
        assert (
            actual_memory_mb >= expected_memory_mb
        ), f"Expected {expected_memory_mb} RAM or greather, got {actual_memory_mb} for node {node}"

        # Check storage size
        actual_disk_size_str = (
            facts["ansible_facts"]
            .get("ansible_devices", {})
            .get("vda", {})
            .get("size", "0.00 GB")
        )
        actual_disk_size_gb = float(
            actual_disk_size_str.split()[0]  # Convert string onto number
        )
        assert (
            actual_disk_size_gb >= expected_disk_size_gb
        ), f"Expected {expected_disk_size_gb} GB or greather storage size, got {actual_disk_size_gb} for node {node}"

Ожидаемые значения параметров указаны в соответствующих переменных:

  • expected_cores – количество ядер CPU;

  • expected_memory_mb – количество установленной RAM в МБ;

  • expected_disk_size_gb – объем хранилища в ГБ.

Сведения об узлах собираются с помощью вспомогательного кода ansible_facts. Метод items возвращает два списка: названия узлов и словари с фактами. В цикле выполняется обход обоих списков. Фактические значения ожидаемых фактов сравниваются с ожидаемыми. Если какой-то факт не удовлетворяет условию, тест считается не пройденным, и в терминал выводится сообщение об ошибке.

Проверка состояния службы#

Этот тест проверяет состояние веб-сервера NGINX:

  • пакет nginx должен быть установлен;

  • служба nginx должна быть включена и запущена.

test_service_status.py#
def test_nginx_installed_and_running(ansible_adhoc):
    """Ensure NGINX is installed and runned."""

    nodes = ansible_adhoc()

    # Check nginx package status
    result_package = nodes.all.package(name="nginx", state="present")
    for node, result in result_package.items():
        assert result.is_successful, f"nginx package is not installed on node {node}"

    # Check nginx.service status
    result_service = nodes.all.service(name="nginx", state="started", enabled=True)

    for node, result in result_service.items():
        assert (
            result.is_succesful
        ), f"nginx.service is not runned or disabled on node {node}"

С помощью вспомогательного кода ansible_adhoc вызываются модули Ansible package и service.

Модуль package проверяет наличие в системе пакета nginx. Поле is_successful содержит значение True только если пакет nginx установлен (state = "present").

Модуль service проверяет состояние службы nginx. Поле is_successful содержит значение True, только если служба nginx запущена (state="started") и запускается при загрузке ОС (enabled=True).