saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

No hacks when interacting with ansible. Access all available modules, not just internal.

Browse source
This commit is contained in:
Pedro Algarvio 2021-05-20 15:03:23 +01:00 • committed by Gareth J. Greenaway
parent 996f85a3b9
commit 55683944a8
4 changed files with 138 additions and 198 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
tests/pytests/functional/modules/test_ansiblegate.py
View file

@ -1,44 +1,43 @@
import pytest
import salt.modules.ansiblegate as ansiblegate
pytestmark = [
pytest.mark.skipif(ansiblegate.ansible is None, reason="Ansible is not installed"),
pytest.mark.skip_on_windows(reason="Not supported on Windows"),
pytest.mark.skip_if_binaries_missing(
"ansible",
"ansible-doc",
"ansible-playbook",
check_all=True,
reason="ansible is not installed",
),
]
def test_ansible_functions_loaded(modules):
"""
Test that the ansible functions are actually loaded
"""
@pytest.fixture
def ansible_ping_func(modules):
if "ansible.system.ping" in modules:
# we need to go by getattr() because salt's loader will try to find "system" in the dictionary and fail
# The ansible hack injects, in this case, "system.ping" as an attribute to the loaded module
ret = getattr(modules.ansible, "system.ping")()
elif "ansible.ping" in modules:
# Ansible >= 2.10
ret = modules.ansible.ping()
else:
pytest.fail("Where is the ping function these days in Ansible?!")
return getattr(modules.ansible, "system.ping")
ret.pop("timeout", None)
if "ansible.ping" in modules:
# Ansible >= 2.10
return modules.ansible.ping
pytest.fail("Where is the ping function these days in Ansible?!")
def test_ansible_functions_loaded(ansible_ping_func):
"""
Test that the ansible functions are actually loaded
"""
ret = ansible_ping_func()
assert ret == {"ping": "pong"}
def test_passing_data_to_ansible_modules(modules):
def test_passing_data_to_ansible_modules(ansible_ping_func):
"""
Test that the ansible functions are actually loaded
"""
expected = "foobar"
if "ansible.system.ping" in modules:
# we need to go by getattr() because salt's loader will try to find "system" in the dictionary and fail
# The ansible hack injects, in this case, "system.ping" as an attribute to the loaded module
ret = getattr(modules.ansible, "system.ping")(data=expected)
elif "ansible.ping" in modules:
# Ansible >= 2.10
ret = modules.ansible.ping(data=expected)
else:
pytest.fail("Where is the ping function these days in Ansible?!")
ret.pop("timeout", None)
ret = ansible_ping_func(data=expected)
assert ret == {"ping": expected}

11
tests/pytests/integration/states/test_ansiblegate.py
View file

@ -17,16 +17,13 @@ pytestmark = [
# Because of the above, these are also destructive tests
pytest.mark.destructive_test,
pytest.mark.skip_if_binaries_missing(
"ansible-playbook", message="ansible-playbook is not installed"
"ansible-playbook", reason="ansible-playbook is not installed"
),
]
@pytest.fixture(scope="module")
def ansible_inventory_directory(tmp_path_factory, grains):
import pprint
pprint.pprint(grains)
if grains["os_family"] != "RedHat":
pytest.skip("Currently, the test targets the RedHat OS familly only.")
tmp_dir = tmp_path_factory.mktemp("ansible")
@ -38,7 +35,7 @@ def ansible_inventory_directory(tmp_path_factory, grains):
@pytest.fixture(scope="module", autouse=True)
def ansible_inventory(ansible_inventory_directory, sshd_server):
inventory = ansible_inventory_directory / "inventory"
inventory = str(ansible_inventory_directory / "inventory")
client_key = str(sshd_server.config_dir / "client_key")
data = {
"all": {
@ -56,9 +53,9 @@ def ansible_inventory(ansible_inventory_directory, sshd_server):
},
},
}
with salt.utils.files.fopen(str(inventory), "w") as yaml_file:
with salt.utils.files.fopen(inventory, "w") as yaml_file:
yaml.dump(data, yaml_file, default_flow_style=False)
return str(inventory)
return inventory
@pytest.mark.requires_sshd_server

230
tests/pytests/unit/modules/test_ansiblegate.py
View file

@ -1,12 +1,9 @@
# Author: Bo Maryniuk <bo@suse.de>
import os
import sys
import pytest
import salt.modules.ansiblegate as ansiblegate
from salt.exceptions import LoaderError
from tests.support.mock import MagicMock, patch
import salt.utils.json
from tests.support.mock import ANY, MagicMock, patch
pytestmark = [
pytest.mark.skip_on_windows(reason="Not supported on Windows"),
@ -18,165 +15,112 @@ def configure_loader_modules():
return {ansiblegate: {}}
@pytest.fixture
def resolver():
_resolver = ansiblegate.AnsibleModuleResolver({})
_resolver._modules_map = {
"one.two.three": os.sep + os.path.join("one", "two", "three.py"),
"four.five.six": os.sep + os.path.join("four", "five", "six.py"),
"three.six.one": os.sep + os.path.join("three", "six", "one.py"),
}
return _resolver
def test_ansible_module_help(resolver):
def test_ansible_module_help():
"""
Test help extraction from the module
:return:
"""
extension = {
"foo": {
"doc": {"description": "The description of foo"},
"examples": "These are the examples",
"return": {"a": "A return"},
}
}
class Module:
"""
An ansible module mock.
"""
__name__ = "foo"
DOCUMENTATION = """
---
one:
text here
---
two:
text here
description:
describe the second part
"""
with patch.object(ansiblegate, "_resolver", resolver), patch.object(
ansiblegate._resolver, "load_module", MagicMock(return_value=Module())
):
ret = ansiblegate.help("dummy")
assert sorted(
ret.get('Available sections on module "{}"'.format(Module().__name__))
) == ["one", "two"]
assert ret.get("Description") == "describe the second part"
with patch("subprocess.run") as proc_run_mock:
proc_run_mock.return_value.stdout = salt.utils.json.dumps(extension)
ret = ansiblegate.help("foo")
assert ret["description"] == extension["foo"]["doc"]["description"]
def test_module_resolver_modlist(resolver):
"""
Test Ansible resolver modules list.
:return:
"""
assert resolver.get_modules_list() == [
"four.five.six",
"one.two.three",
"three.six.one",
]
for ptr in ["five", "fi", "ve"]:
assert resolver.get_modules_list(ptr) == ["four.five.six"]
for ptr in ["si", "ix", "six"]:
assert resolver.get_modules_list(ptr) == ["four.five.six", "three.six.one"]
assert resolver.get_modules_list("one") == ["one.two.three", "three.six.one"]
assert resolver.get_modules_list("one.two") == ["one.two.three"]
assert resolver.get_modules_list("four") == ["four.five.six"]
def test_resolver_module_loader_failure(resolver):
"""
Test Ansible module loader.
:return:
"""
mod = "four.five.six"
with pytest.raises(ImportError):
resolver.load_module(mod)
mod = "i.even.do.not.exist.at.all"
with pytest.raises(LoaderError):
resolver.load_module(mod)
def test_resolver_module_loader(resolver):
"""
Test Ansible module loader.
:return:
"""
with patch("salt.modules.ansiblegate.importlib", MagicMock()), patch(
"salt.modules.ansiblegate.importlib.import_module", lambda x: x
):
assert resolver.load_module("four.five.six") == "ansible.modules.four.five.six"
def test_resolver_module_loader_import_failure(resolver):
"""
Test Ansible module loader failure.
:return:
"""
with patch("salt.modules.ansiblegate.importlib", MagicMock()), patch(
"salt.modules.ansiblegate.importlib.import_module", lambda x: x
):
with pytest.raises(LoaderError):
resolver.load_module("something.strange")
def test_virtual_function(resolver):
def test_virtual_function(subtests):
"""
Test Ansible module __virtual__ when ansible is not installed on the minion.
:return:
"""
with patch("salt.modules.ansiblegate.ansible", None):
assert ansiblegate.__virtual__() == (
False,
"Ansible is not installed on this system",
)
with subtests.test("missing ansible binary"):
with patch("salt.utils.path.which", side_effect=[None]):
assert ansiblegate.__virtual__() == (
False,
"The 'ansible' binary was not found.",
)
with subtests.test("missing ansible-doc binary"):
with patch(
"salt.utils.path.which", side_effect=["/path/to/ansible", None],
):
assert ansiblegate.__virtual__() == (
False,
"The 'ansible-doc' binary was not found.",
)
with subtests.test("missing ansible-playbook binary"):
with patch(
"salt.utils.path.which",
side_effect=["/path/to/ansible", "/path/to/ansible-doc", None],
):
assert ansiblegate.__virtual__() == (
False,
"The 'ansible-playbook' binary was not found.",
)
with subtests.test("Failing to load the ansible modules listing"):
with patch(
"salt.utils.path.which",
side_effect=[
"/path/to/ansible",
"/path/to/ansible-doc",
"/path/to/ansible-playbook",
],
):
with patch("subprocess.run") as proc_run_mock:
proc_run_mock.return_value.retcode = 1
proc_run_mock.return_value.stderr = "bar"
proc_run_mock.return_value.stdout = "{}"
assert ansiblegate.__virtual__() == (
False,
"Failed to get the listing of ansible modules:\nbar",
)
@pytest.mark.skipif(
sys.version_info < (3, 6),
reason="Skipped on Py3.5, the mock of subprocess.run is different",
)
def test_ansible_module_call(resolver):
def test_ansible_module_call():
"""
Test Ansible module call from ansible gate module
:return:
"""
class Module:
"""
An ansible module mock.
"""
with patch("subprocess.run") as proc_run_mock:
proc_run_mock.return_value.stdout = (
'localhost | SUCCESS => {\n "completed": true \n}'
)
__name__ = "one.two.three"
__file__ = "foofile"
def main(): # pylint: disable=no-method-argument
pass
with patch.object(ansiblegate, "_resolver", resolver), patch.object(
ansiblegate._resolver, "load_module", MagicMock(return_value=Module())
):
_ansible_module_caller = ansiblegate.AnsibleModuleCaller(ansiblegate._resolver)
with patch("subprocess.run") as proc_run_mock:
proc_run_mock.return_value.stdout = '{"completed": true}'
ret = _ansible_module_caller.call("one.two.three", "arg_1", kwarg1="foobar")
proc_run_mock.assert_any_call(
[
sys.executable,
"-c",
"import sys, one.two.three; print(one.two.three.main(), file=sys.stdout); sys.stdout.flush()",
],
input='{"ANSIBLE_MODULE_ARGS": {"kwarg1": "foobar", "_raw_params": "arg_1"}}',
stdout=-1,
stderr=-1,
check=True,
shell=False,
universal_newlines=True,
timeout=1200,
)
assert ret == {"completed": True, "timeout": 1200}
ret = ansiblegate.call("one.two.three", "arg_1", kwarg1="foobar")
proc_run_mock.assert_any_call(
[
ANY,
"localhost",
"--limit",
"127.0.0.1",
"-m",
"one.two.three",
"-a",
'"arg_1" kwarg1="foobar"',
"-i",
ANY,
],
check=True,
shell=False,
stderr=-1,
stdout=-1,
timeout=1200,
universal_newlines=True,
)
assert ret == {"completed": True}
def test_ansible_playbooks_return_retcode(resolver):
def test_ansible_playbooks_return_retcode():
"""
Test ansible.playbooks execution module function include retcode in the return.
:return:

46
tests/pytests/unit/states/test_ansiblegate.py
View file

@ -29,16 +29,17 @@ def test_ansible_playbooks_states_success(playbooks_examples_dir):
with patch.dict(
ansiblegate.__salt__,
{"ansible.playbooks": MagicMock(return_value=success_output)},
), patch("salt.utils.path.which", MagicMock(return_value=True)):
with patch.dict(ansiblegate.__opts__, {"test": False}):
ret = ansiblegate.playbooks("foobar")
assert ret["result"] is True
assert ret["comment"] == "Changes were made by playbook foobar"
assert ret["changes"] == {
"py2hosts": {
"Ansible copy file to remote server": {"centos7-host1.tf.local": {}}
}
), patch("salt.utils.path.which", return_value=True), patch.dict(
ansiblegate.__opts__, {"test": False}
):
ret = ansiblegate.playbooks("foobar")
assert ret["result"] is True
assert ret["comment"] == "Changes were made by playbook foobar"
assert ret["changes"] == {
"py2hosts": {
"Ansible copy file to remote server": {"centos7-host1.tf.local": {}}
}
}
def test_ansible_playbooks_states_failed(playbooks_examples_dir):
@ -52,19 +53,18 @@ def test_ansible_playbooks_states_failed(playbooks_examples_dir):
with patch.dict(
ansiblegate.__salt__,
{"ansible.playbooks": MagicMock(return_value=failed_output)},
), patch("salt.utils.path.which", MagicMock(return_value=True)):
with patch.dict(ansiblegate.__opts__, {"test": False}):
ret = ansiblegate.playbooks("foobar")
assert ret["result"] is False
assert (
ret["comment"] == "There were some issues running the playbook foobar"
)
assert ret["changes"] == {
"py2hosts": {
"yum": {
"centos7-host1.tf.local": [
"No package matching 'rsyndc' found available, installed or updated"
]
}
), patch("salt.utils.path.which", return_value=True), patch.dict(
ansiblegate.__opts__, {"test": False}
):
ret = ansiblegate.playbooks("foobar")
assert ret["result"] is False
assert ret["comment"] == "There were some issues running the playbook foobar"
assert ret["changes"] == {
"py2hosts": {
"yum": {
"centos7-host1.tf.local": [
"No package matching 'rsyndc' found available, installed or updated"
]
}
}
}