mirror of
https://github.com/freeipa/ansible-freeipa.git
synced 2026-07-31 20:04:45 +00:00
Added ability to add pytest tests
Until now ansible-freeipa repository only had playbook tests. This commit introduces the ability of creating TestCase classes connected to the master host. This connection can be used to run commands in the managed host after the ansible playbook execution is the allowing the verification of the machine state.
This commit is contained in:
Sergio Oliveira Campos
@@ -1,117 +1,38 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
import os
|
||||
import functools
|
||||
import tempfile
|
||||
# Authors:
|
||||
# Sergio Oliveira Campos <seocam@redhat.com>
|
||||
#
|
||||
# Copyright (C) 2020 Red Hat
|
||||
# see file 'COPYING' for use and warranty information
|
||||
#
|
||||
# This program is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation, either version 3 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import subprocess
|
||||
import pytest
|
||||
import functools
|
||||
|
||||
from unittest import TestCase
|
||||
|
||||
import pytest
|
||||
|
||||
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
|
||||
|
||||
def is_docker_env():
|
||||
if os.getenv("RUN_TESTS_IN_DOCKER", "0") == "0":
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def get_ssh_password():
|
||||
return os.getenv("IPA_SSH_PASSWORD")
|
||||
|
||||
|
||||
def get_server_host():
|
||||
return os.getenv("IPA_SERVER_HOST")
|
||||
|
||||
|
||||
def get_molecule_scenario():
|
||||
return get_server_host()
|
||||
|
||||
|
||||
def get_inventory_content():
|
||||
ipa_server_host = get_server_host()
|
||||
|
||||
if is_docker_env():
|
||||
ipa_server_host += " ansible_connection=docker"
|
||||
|
||||
sshpass = get_ssh_password()
|
||||
if sshpass:
|
||||
ipa_server_host += " ansible_ssh_pass=%s" % sshpass
|
||||
|
||||
lines = [
|
||||
"[ipaserver]",
|
||||
ipa_server_host,
|
||||
"[ipaserver:vars]",
|
||||
"ipaserver_domain=test.local",
|
||||
"ipaserver_realm=TEST.LOCAL",
|
||||
]
|
||||
return "\n".join(lines).encode("utf8")
|
||||
|
||||
|
||||
def write_logs(result, test_name):
|
||||
log_dir = os.path.join(SCRIPT_DIR, "logs")
|
||||
if not os.path.exists(log_dir):
|
||||
os.makedirs(log_dir)
|
||||
|
||||
# Write stdout log for test
|
||||
log_path = os.path.join(log_dir, test_name + ".log")
|
||||
with open(log_path, "w") as log_file:
|
||||
log_file.write(result.stdout.decode("utf-8"))
|
||||
|
||||
# Write stderr log for test
|
||||
error_log_path = os.path.join(log_dir, test_name + "-error.log")
|
||||
with open(error_log_path, "w") as log_file:
|
||||
log_file.write(result.stderr.decode("utf-8"))
|
||||
|
||||
|
||||
def run_playbook(playbook, test_name):
|
||||
with tempfile.NamedTemporaryFile() as inventory_file:
|
||||
inventory_file.write(get_inventory_content())
|
||||
inventory_file.flush()
|
||||
cmd = [
|
||||
"ansible-playbook",
|
||||
"-i",
|
||||
inventory_file.name,
|
||||
playbook,
|
||||
]
|
||||
process = subprocess.run(
|
||||
cmd, cwd=SCRIPT_DIR, stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
||||
)
|
||||
write_logs(process, test_name)
|
||||
|
||||
return process
|
||||
|
||||
|
||||
def list_test_yaml(dir_path):
|
||||
yamls = []
|
||||
for yaml_name in sorted(os.listdir(dir_path)):
|
||||
if yaml_name.startswith("test_") and yaml_name.endswith(".yml"):
|
||||
yamls.append(
|
||||
{
|
||||
"path": os.path.join(dir_path, yaml_name),
|
||||
"name": yaml_name.split(".")[0],
|
||||
}
|
||||
)
|
||||
return yamls
|
||||
|
||||
|
||||
def get_test_groups():
|
||||
test_dirs = os.listdir(SCRIPT_DIR)
|
||||
groups = {}
|
||||
for test_group_dir in sorted(test_dirs):
|
||||
group_dir_path = os.path.join(SCRIPT_DIR, test_group_dir)
|
||||
if not os.path.isdir(group_dir_path):
|
||||
continue
|
||||
yamls = list_test_yaml(group_dir_path)
|
||||
if yamls:
|
||||
groups[test_group_dir] = yamls
|
||||
return groups
|
||||
from utils import get_test_playbooks, get_server_host, run_playbook
|
||||
|
||||
|
||||
def prepare_test(test_name, test_path):
|
||||
"""Decorator for the tests generated automatically from playbooks.
|
||||
|
||||
Injects 2 arguments to the test (`test_path` and `test_name`) and
|
||||
name the test method using test name (to ensure test reports are useful).
|
||||
"""
|
||||
def decorator(func):
|
||||
@functools.wraps(func)
|
||||
def wrapper(*args, **kwargs):
|
||||
@@ -127,36 +48,20 @@ def prepare_test(test_name, test_path):
|
||||
|
||||
# Dynamically create the TestCase classes with respective
|
||||
# test_* methods.
|
||||
for group_name, group_tests in get_test_groups().items():
|
||||
for test_dir_name, playbooks_in_dir in get_test_playbooks().items():
|
||||
_tests = {}
|
||||
for test_config in group_tests:
|
||||
test_name = test_config["name"].replace("-", "_")
|
||||
test_path = test_config["path"]
|
||||
for playbook in playbooks_in_dir:
|
||||
test_name = playbook["name"].replace("-", "_")
|
||||
test_path = playbook["path"]
|
||||
|
||||
@pytest.mark.skipif(
|
||||
not get_server_host(),
|
||||
reason="Environment variable IPA_SERVER_HOST must be set",
|
||||
)
|
||||
@pytest.mark.playbook
|
||||
@prepare_test(test_name, test_path)
|
||||
def method(self, test_path, test_name):
|
||||
result = run_playbook(test_path, test_name)
|
||||
status_code_msg = "ansible-playbook return code: {}".format(
|
||||
result.returncode
|
||||
)
|
||||
assert_msg = "\n".join(
|
||||
[
|
||||
"",
|
||||
"-" * 30 + " Captured stdout " + "-" * 30,
|
||||
result.stdout.decode("utf8"),
|
||||
"-" * 30 + " Captured stderr " + "-" * 30,
|
||||
result.stderr.decode("utf8"),
|
||||
"-" * 30 + " Playbook Return Code " + "-" * 30,
|
||||
status_code_msg,
|
||||
]
|
||||
)
|
||||
# Need to get the last bytes of msg otherwise Azure
|
||||
# will cut it out.
|
||||
assert result.returncode == 0, assert_msg[-2500:]
|
||||
run_playbook(test_path)
|
||||
|
||||
_tests[test_name] = method
|
||||
globals()[group_name] = type(group_name, tuple([TestCase]), _tests,)
|
||||
globals()[test_dir_name] = type(test_dir_name, tuple([TestCase]), _tests,)
|
||||
|
||||
Reference in New Issue
Block a user