summaryrefslogtreecommitdiffstats
path: root/test/units/executor
diff options
context:
space:
mode:
Diffstat (limited to 'test/units/executor')
-rw-r--r--test/units/executor/__init__.py0
-rw-r--r--test/units/executor/module_common/test_modify_module.py43
-rw-r--r--test/units/executor/module_common/test_module_common.py200
-rw-r--r--test/units/executor/module_common/test_recursive_finder.py130
-rw-r--r--test/units/executor/test_interpreter_discovery.py86
-rw-r--r--test/units/executor/test_play_iterator.py462
-rw-r--r--test/units/executor/test_playbook_executor.py148
-rw-r--r--test/units/executor/test_task_executor.py489
-rw-r--r--test/units/executor/test_task_queue_manager_callbacks.py121
-rw-r--r--test/units/executor/test_task_result.py171
10 files changed, 1850 insertions, 0 deletions
diff --git a/test/units/executor/__init__.py b/test/units/executor/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/test/units/executor/__init__.py
diff --git a/test/units/executor/module_common/test_modify_module.py b/test/units/executor/module_common/test_modify_module.py
new file mode 100644
index 0000000..dceef76
--- /dev/null
+++ b/test/units/executor/module_common/test_modify_module.py
@@ -0,0 +1,43 @@
+# Copyright (c) 2018 Matt Martz <matt@sivel.net>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+# -*- coding: utf-8 -*-
+
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import pytest
+
+from ansible.executor.module_common import modify_module
+from ansible.module_utils.six import PY2
+
+from test_module_common import templar
+
+
+FAKE_OLD_MODULE = b'''#!/usr/bin/python
+import sys
+print('{"result": "%s"}' % sys.executable)
+'''
+
+
+@pytest.fixture
+def fake_old_module_open(mocker):
+ m = mocker.mock_open(read_data=FAKE_OLD_MODULE)
+ if PY2:
+ mocker.patch('__builtin__.open', m)
+ else:
+ mocker.patch('builtins.open', m)
+
+# this test no longer makes sense, since a Python module will always either have interpreter discovery run or
+# an explicit interpreter passed (so we'll never default to the module shebang)
+# def test_shebang(fake_old_module_open, templar):
+# (data, style, shebang) = modify_module('fake_module', 'fake_path', {}, templar)
+# assert shebang == '#!/usr/bin/python'
+
+
+def test_shebang_task_vars(fake_old_module_open, templar):
+ task_vars = {
+ 'ansible_python_interpreter': '/usr/bin/python3'
+ }
+
+ (data, style, shebang) = modify_module('fake_module', 'fake_path', {}, templar, task_vars=task_vars)
+ assert shebang == '#!/usr/bin/python3'
diff --git a/test/units/executor/module_common/test_module_common.py b/test/units/executor/module_common/test_module_common.py
new file mode 100644
index 0000000..fa6add8
--- /dev/null
+++ b/test/units/executor/module_common/test_module_common.py
@@ -0,0 +1,200 @@
+# (c) 2017, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os.path
+
+import pytest
+
+import ansible.errors
+
+from ansible.executor import module_common as amc
+from ansible.executor.interpreter_discovery import InterpreterDiscoveryRequiredError
+from ansible.module_utils.six import PY2
+
+
+class TestStripComments:
+ def test_no_changes(self):
+ no_comments = u"""def some_code():
+ return False"""
+ assert amc._strip_comments(no_comments) == no_comments
+
+ def test_all_comments(self):
+ all_comments = u"""# This is a test
+ # Being as it is
+ # To be
+ """
+ assert amc._strip_comments(all_comments) == u""
+
+ def test_all_whitespace(self):
+ # Note: Do not remove the spaces on the blank lines below. They're
+ # test data to show that the lines get removed despite having spaces
+ # on them
+ all_whitespace = u"""
+
+
+
+\t\t\r\n
+ """ # nopep8
+ assert amc._strip_comments(all_whitespace) == u""
+
+ def test_somewhat_normal(self):
+ mixed = u"""#!/usr/bin/python
+
+# here we go
+def test(arg):
+ # this is a thing
+ thing = '# test'
+ return thing
+# End
+"""
+ mixed_results = u"""def test(arg):
+ thing = '# test'
+ return thing"""
+ assert amc._strip_comments(mixed) == mixed_results
+
+
+class TestSlurp:
+ def test_slurp_nonexistent(self, mocker):
+ mocker.patch('os.path.exists', side_effect=lambda x: False)
+ with pytest.raises(ansible.errors.AnsibleError):
+ amc._slurp('no_file')
+
+ def test_slurp_file(self, mocker):
+ mocker.patch('os.path.exists', side_effect=lambda x: True)
+ m = mocker.mock_open(read_data='This is a test')
+ if PY2:
+ mocker.patch('__builtin__.open', m)
+ else:
+ mocker.patch('builtins.open', m)
+ assert amc._slurp('some_file') == 'This is a test'
+
+ def test_slurp_file_with_newlines(self, mocker):
+ mocker.patch('os.path.exists', side_effect=lambda x: True)
+ m = mocker.mock_open(read_data='#!/usr/bin/python\ndef test(args):\nprint("hi")\n')
+ if PY2:
+ mocker.patch('__builtin__.open', m)
+ else:
+ mocker.patch('builtins.open', m)
+ assert amc._slurp('some_file') == '#!/usr/bin/python\ndef test(args):\nprint("hi")\n'
+
+
+@pytest.fixture
+def templar():
+ class FakeTemplar:
+ def template(self, template_string, *args, **kwargs):
+ return template_string
+
+ return FakeTemplar()
+
+
+class TestGetShebang:
+ """Note: We may want to change the API of this function in the future. It isn't a great API"""
+ def test_no_interpreter_set(self, templar):
+ # normally this would return /usr/bin/python, but so long as we're defaulting to auto python discovery, we'll get
+ # an InterpreterDiscoveryRequiredError here instead
+ with pytest.raises(InterpreterDiscoveryRequiredError):
+ amc._get_shebang(u'/usr/bin/python', {}, templar)
+
+ def test_python_interpreter(self, templar):
+ assert amc._get_shebang(u'/usr/bin/python3.8', {}, templar) == ('#!/usr/bin/python3.8', u'/usr/bin/python3.8')
+
+ def test_non_python_interpreter(self, templar):
+ assert amc._get_shebang(u'/usr/bin/ruby', {}, templar) == ('#!/usr/bin/ruby', u'/usr/bin/ruby')
+
+ def test_interpreter_set_in_task_vars(self, templar):
+ assert amc._get_shebang(u'/usr/bin/python', {u'ansible_python_interpreter': u'/usr/bin/pypy'}, templar) == \
+ (u'#!/usr/bin/pypy', u'/usr/bin/pypy')
+
+ def test_non_python_interpreter_in_task_vars(self, templar):
+ assert amc._get_shebang(u'/usr/bin/ruby', {u'ansible_ruby_interpreter': u'/usr/local/bin/ruby'}, templar) == \
+ (u'#!/usr/local/bin/ruby', u'/usr/local/bin/ruby')
+
+ def test_with_args(self, templar):
+ assert amc._get_shebang(u'/usr/bin/python', {u'ansible_python_interpreter': u'/usr/bin/python3'}, templar, args=('-tt', '-OO')) == \
+ (u'#!/usr/bin/python3 -tt -OO', u'/usr/bin/python3')
+
+ def test_python_via_env(self, templar):
+ assert amc._get_shebang(u'/usr/bin/python', {u'ansible_python_interpreter': u'/usr/bin/env python'}, templar) == \
+ (u'#!/usr/bin/env python', u'/usr/bin/env python')
+
+
+class TestDetectionRegexes:
+ ANSIBLE_MODULE_UTIL_STRINGS = (
+ # Absolute collection imports
+ b'import ansible_collections.my_ns.my_col.plugins.module_utils.my_util',
+ b'from ansible_collections.my_ns.my_col.plugins.module_utils import my_util',
+ b'from ansible_collections.my_ns.my_col.plugins.module_utils.my_util import my_func',
+ # Absolute core imports
+ b'import ansible.module_utils.basic',
+ b'from ansible.module_utils import basic',
+ b'from ansible.module_utils.basic import AnsibleModule',
+ # Relative imports
+ b'from ..module_utils import basic',
+ b'from .. module_utils import basic',
+ b'from ....module_utils import basic',
+ b'from ..module_utils.basic import AnsibleModule',
+ )
+ NOT_ANSIBLE_MODULE_UTIL_STRINGS = (
+ b'from ansible import release',
+ b'from ..release import __version__',
+ b'from .. import release',
+ b'from ansible.modules.system import ping',
+ b'from ansible_collecitons.my_ns.my_col.plugins.modules import function',
+ )
+
+ OFFSET = os.path.dirname(os.path.dirname(amc.__file__))
+ CORE_PATHS = (
+ ('%s/modules/from_role.py' % OFFSET, 'ansible/modules/from_role'),
+ ('%s/modules/system/ping.py' % OFFSET, 'ansible/modules/system/ping'),
+ ('%s/modules/cloud/amazon/s3.py' % OFFSET, 'ansible/modules/cloud/amazon/s3'),
+ )
+
+ COLLECTION_PATHS = (
+ ('/root/ansible_collections/ns/col/plugins/modules/ping.py',
+ 'ansible_collections/ns/col/plugins/modules/ping'),
+ ('/root/ansible_collections/ns/col/plugins/modules/subdir/ping.py',
+ 'ansible_collections/ns/col/plugins/modules/subdir/ping'),
+ )
+
+ @pytest.mark.parametrize('testcase', ANSIBLE_MODULE_UTIL_STRINGS)
+ def test_detect_new_style_python_module_re(self, testcase):
+ assert amc.NEW_STYLE_PYTHON_MODULE_RE.search(testcase)
+
+ @pytest.mark.parametrize('testcase', NOT_ANSIBLE_MODULE_UTIL_STRINGS)
+ def test_no_detect_new_style_python_module_re(self, testcase):
+ assert not amc.NEW_STYLE_PYTHON_MODULE_RE.search(testcase)
+
+ # pylint bug: https://github.com/PyCQA/pylint/issues/511
+ @pytest.mark.parametrize('testcase, result', CORE_PATHS) # pylint: disable=undefined-variable
+ def test_detect_core_library_path_re(self, testcase, result):
+ assert amc.CORE_LIBRARY_PATH_RE.search(testcase).group('path') == result
+
+ @pytest.mark.parametrize('testcase', (p[0] for p in COLLECTION_PATHS)) # pylint: disable=undefined-variable
+ def test_no_detect_core_library_path_re(self, testcase):
+ assert not amc.CORE_LIBRARY_PATH_RE.search(testcase)
+
+ @pytest.mark.parametrize('testcase, result', COLLECTION_PATHS) # pylint: disable=undefined-variable
+ def test_detect_collection_path_re(self, testcase, result):
+ assert amc.COLLECTION_PATH_RE.search(testcase).group('path') == result
+
+ @pytest.mark.parametrize('testcase', (p[0] for p in CORE_PATHS)) # pylint: disable=undefined-variable
+ def test_no_detect_collection_path_re(self, testcase):
+ assert not amc.COLLECTION_PATH_RE.search(testcase)
diff --git a/test/units/executor/module_common/test_recursive_finder.py b/test/units/executor/module_common/test_recursive_finder.py
new file mode 100644
index 0000000..8136a00
--- /dev/null
+++ b/test/units/executor/module_common/test_recursive_finder.py
@@ -0,0 +1,130 @@
+# (c) 2017, Toshio Kuratomi <tkuratomi@ansible.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+import os
+import pytest
+import zipfile
+
+from collections import namedtuple
+from io import BytesIO
+
+import ansible.errors
+
+from ansible.executor.module_common import recursive_finder
+
+
+# These are the modules that are brought in by module_utils/basic.py This may need to be updated
+# when basic.py gains new imports
+# We will remove these when we modify AnsiBallZ to store its args in a separate file instead of in
+# basic.py
+
+MODULE_UTILS_BASIC_FILES = frozenset(('ansible/__init__.py',
+ 'ansible/module_utils/__init__.py',
+ 'ansible/module_utils/_text.py',
+ 'ansible/module_utils/basic.py',
+ 'ansible/module_utils/six/__init__.py',
+ 'ansible/module_utils/_text.py',
+ 'ansible/module_utils/common/_collections_compat.py',
+ 'ansible/module_utils/common/_json_compat.py',
+ 'ansible/module_utils/common/collections.py',
+ 'ansible/module_utils/common/parameters.py',
+ 'ansible/module_utils/common/warnings.py',
+ 'ansible/module_utils/parsing/convert_bool.py',
+ 'ansible/module_utils/common/__init__.py',
+ 'ansible/module_utils/common/file.py',
+ 'ansible/module_utils/common/locale.py',
+ 'ansible/module_utils/common/process.py',
+ 'ansible/module_utils/common/sys_info.py',
+ 'ansible/module_utils/common/text/__init__.py',
+ 'ansible/module_utils/common/text/converters.py',
+ 'ansible/module_utils/common/text/formatters.py',
+ 'ansible/module_utils/common/validation.py',
+ 'ansible/module_utils/common/_utils.py',
+ 'ansible/module_utils/common/arg_spec.py',
+ 'ansible/module_utils/compat/__init__.py',
+ 'ansible/module_utils/compat/_selectors2.py',
+ 'ansible/module_utils/compat/selectors.py',
+ 'ansible/module_utils/compat/selinux.py',
+ 'ansible/module_utils/distro/__init__.py',
+ 'ansible/module_utils/distro/_distro.py',
+ 'ansible/module_utils/errors.py',
+ 'ansible/module_utils/parsing/__init__.py',
+ 'ansible/module_utils/parsing/convert_bool.py',
+ 'ansible/module_utils/pycompat24.py',
+ 'ansible/module_utils/six/__init__.py',
+ ))
+
+ONLY_BASIC_FILE = frozenset(('ansible/module_utils/basic.py',))
+
+ANSIBLE_LIB = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))), 'lib', 'ansible')
+
+
+@pytest.fixture
+def finder_containers():
+ FinderContainers = namedtuple('FinderContainers', ['zf'])
+
+ zipoutput = BytesIO()
+ zf = zipfile.ZipFile(zipoutput, mode='w', compression=zipfile.ZIP_STORED)
+
+ return FinderContainers(zf)
+
+
+class TestRecursiveFinder(object):
+ def test_no_module_utils(self, finder_containers):
+ name = 'ping'
+ data = b'#!/usr/bin/python\nreturn \'{\"changed\": false}\''
+ recursive_finder(name, os.path.join(ANSIBLE_LIB, 'modules', 'system', 'ping.py'), data, *finder_containers)
+ assert frozenset(finder_containers.zf.namelist()) == MODULE_UTILS_BASIC_FILES
+
+ def test_module_utils_with_syntax_error(self, finder_containers):
+ name = 'fake_module'
+ data = b'#!/usr/bin/python\ndef something(:\n pass\n'
+ with pytest.raises(ansible.errors.AnsibleError) as exec_info:
+ recursive_finder(name, os.path.join(ANSIBLE_LIB, 'modules', 'system', 'fake_module.py'), data, *finder_containers)
+ assert 'Unable to import fake_module due to invalid syntax' in str(exec_info.value)
+
+ def test_module_utils_with_identation_error(self, finder_containers):
+ name = 'fake_module'
+ data = b'#!/usr/bin/python\n def something():\n pass\n'
+ with pytest.raises(ansible.errors.AnsibleError) as exec_info:
+ recursive_finder(name, os.path.join(ANSIBLE_LIB, 'modules', 'system', 'fake_module.py'), data, *finder_containers)
+ assert 'Unable to import fake_module due to unexpected indent' in str(exec_info.value)
+
+ #
+ # Test importing six with many permutations because it is not a normal module
+ #
+ def test_from_import_six(self, finder_containers):
+ name = 'ping'
+ data = b'#!/usr/bin/python\nfrom ansible.module_utils import six'
+ recursive_finder(name, os.path.join(ANSIBLE_LIB, 'modules', 'system', 'ping.py'), data, *finder_containers)
+ assert frozenset(finder_containers.zf.namelist()) == frozenset(('ansible/module_utils/six/__init__.py', )).union(MODULE_UTILS_BASIC_FILES)
+
+ def test_import_six(self, finder_containers):
+ name = 'ping'
+ data = b'#!/usr/bin/python\nimport ansible.module_utils.six'
+ recursive_finder(name, os.path.join(ANSIBLE_LIB, 'modules', 'system', 'ping.py'), data, *finder_containers)
+ assert frozenset(finder_containers.zf.namelist()) == frozenset(('ansible/module_utils/six/__init__.py', )).union(MODULE_UTILS_BASIC_FILES)
+
+ def test_import_six_from_many_submodules(self, finder_containers):
+ name = 'ping'
+ data = b'#!/usr/bin/python\nfrom ansible.module_utils.six.moves.urllib.parse import urlparse'
+ recursive_finder(name, os.path.join(ANSIBLE_LIB, 'modules', 'system', 'ping.py'), data, *finder_containers)
+ assert frozenset(finder_containers.zf.namelist()) == frozenset(('ansible/module_utils/six/__init__.py',)).union(MODULE_UTILS_BASIC_FILES)
diff --git a/test/units/executor/test_interpreter_discovery.py b/test/units/executor/test_interpreter_discovery.py
new file mode 100644
index 0000000..43db595
--- /dev/null
+++ b/test/units/executor/test_interpreter_discovery.py
@@ -0,0 +1,86 @@
+# -*- coding: utf-8 -*-
+# (c) 2019, Jordan Borean <jborean@redhat.com>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from unittest.mock import MagicMock
+
+from ansible.executor.interpreter_discovery import discover_interpreter
+from ansible.module_utils._text import to_text
+
+mock_ubuntu_platform_res = to_text(
+ r'{"osrelease_content": "NAME=\"Ubuntu\"\nVERSION=\"16.04.5 LTS (Xenial Xerus)\"\nID=ubuntu\nID_LIKE=debian\n'
+ r'PRETTY_NAME=\"Ubuntu 16.04.5 LTS\"\nVERSION_ID=\"16.04\"\nHOME_URL=\"http://www.ubuntu.com/\"\n'
+ r'SUPPORT_URL=\"http://help.ubuntu.com/\"\nBUG_REPORT_URL=\"http://bugs.launchpad.net/ubuntu/\"\n'
+ r'VERSION_CODENAME=xenial\nUBUNTU_CODENAME=xenial\n", "platform_dist_result": ["Ubuntu", "16.04", "xenial"]}'
+)
+
+
+def test_discovery_interpreter_linux_auto_legacy():
+ res1 = u'PLATFORM\nLinux\nFOUND\n/usr/bin/python\n/usr/bin/python3.5\n/usr/bin/python3\nENDFOUND'
+
+ mock_action = MagicMock()
+ mock_action._low_level_execute_command.side_effect = [{'stdout': res1}, {'stdout': mock_ubuntu_platform_res}]
+
+ actual = discover_interpreter(mock_action, 'python', 'auto_legacy', {'inventory_hostname': u'host-fóöbär'})
+
+ assert actual == u'/usr/bin/python'
+ assert len(mock_action.method_calls) == 3
+ assert mock_action.method_calls[2][0] == '_discovery_warnings.append'
+ assert u'Distribution Ubuntu 16.04 on host host-fóöbär should use /usr/bin/python3, but is using /usr/bin/python' \
+ u' for backward compatibility' in mock_action.method_calls[2][1][0]
+
+
+def test_discovery_interpreter_linux_auto_legacy_silent():
+ res1 = u'PLATFORM\nLinux\nFOUND\n/usr/bin/python\n/usr/bin/python3.5\n/usr/bin/python3\nENDFOUND'
+
+ mock_action = MagicMock()
+ mock_action._low_level_execute_command.side_effect = [{'stdout': res1}, {'stdout': mock_ubuntu_platform_res}]
+
+ actual = discover_interpreter(mock_action, 'python', 'auto_legacy_silent', {'inventory_hostname': u'host-fóöbär'})
+
+ assert actual == u'/usr/bin/python'
+ assert len(mock_action.method_calls) == 2
+
+
+def test_discovery_interpreter_linux_auto():
+ res1 = u'PLATFORM\nLinux\nFOUND\n/usr/bin/python\n/usr/bin/python3.5\n/usr/bin/python3\nENDFOUND'
+
+ mock_action = MagicMock()
+ mock_action._low_level_execute_command.side_effect = [{'stdout': res1}, {'stdout': mock_ubuntu_platform_res}]
+
+ actual = discover_interpreter(mock_action, 'python', 'auto', {'inventory_hostname': u'host-fóöbär'})
+
+ assert actual == u'/usr/bin/python3'
+ assert len(mock_action.method_calls) == 2
+
+
+def test_discovery_interpreter_non_linux():
+ mock_action = MagicMock()
+ mock_action._low_level_execute_command.return_value = \
+ {'stdout': u'PLATFORM\nDarwin\nFOUND\n/usr/bin/python\nENDFOUND'}
+
+ actual = discover_interpreter(mock_action, 'python', 'auto_legacy', {'inventory_hostname': u'host-fóöbär'})
+
+ assert actual == u'/usr/bin/python'
+ assert len(mock_action.method_calls) == 2
+ assert mock_action.method_calls[1][0] == '_discovery_warnings.append'
+ assert u'Platform darwin on host host-fóöbär is using the discovered Python interpreter at /usr/bin/python, ' \
+ u'but future installation of another Python interpreter could change the meaning of that path' \
+ in mock_action.method_calls[1][1][0]
+
+
+def test_no_interpreters_found():
+ mock_action = MagicMock()
+ mock_action._low_level_execute_command.return_value = {'stdout': u'PLATFORM\nWindows\nFOUND\nENDFOUND'}
+
+ actual = discover_interpreter(mock_action, 'python', 'auto_legacy', {'inventory_hostname': u'host-fóöbär'})
+
+ assert actual == u'/usr/bin/python'
+ assert len(mock_action.method_calls) == 2
+ assert mock_action.method_calls[1][0] == '_discovery_warnings.append'
+ assert u'No python interpreters found for host host-fóöbär (tried' \
+ in mock_action.method_calls[1][1][0]
diff --git a/test/units/executor/test_play_iterator.py b/test/units/executor/test_play_iterator.py
new file mode 100644
index 0000000..6670888
--- /dev/null
+++ b/test/units/executor/test_play_iterator.py
@@ -0,0 +1,462 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from units.compat import unittest
+from unittest.mock import patch, MagicMock
+
+from ansible.executor.play_iterator import HostState, PlayIterator, IteratingStates, FailedStates
+from ansible.playbook import Playbook
+from ansible.playbook.play_context import PlayContext
+
+from units.mock.loader import DictDataLoader
+from units.mock.path import mock_unfrackpath_noop
+
+
+class TestPlayIterator(unittest.TestCase):
+
+ def test_host_state(self):
+ hs = HostState(blocks=list(range(0, 10)))
+ hs.tasks_child_state = HostState(blocks=[0])
+ hs.rescue_child_state = HostState(blocks=[1])
+ hs.always_child_state = HostState(blocks=[2])
+ repr(hs)
+ hs.run_state = 100
+ repr(hs)
+ hs.fail_state = 15
+ repr(hs)
+
+ for i in range(0, 10):
+ hs.cur_block = i
+ self.assertEqual(hs.get_current_block(), i)
+
+ new_hs = hs.copy()
+
+ @patch('ansible.playbook.role.definition.unfrackpath', mock_unfrackpath_noop)
+ def test_play_iterator(self):
+ fake_loader = DictDataLoader({
+ "test_play.yml": """
+ - hosts: all
+ gather_facts: false
+ roles:
+ - test_role
+ pre_tasks:
+ - debug: msg="this is a pre_task"
+ tasks:
+ - debug: msg="this is a regular task"
+ - block:
+ - debug: msg="this is a block task"
+ - block:
+ - debug: msg="this is a sub-block in a block"
+ rescue:
+ - debug: msg="this is a rescue task"
+ - block:
+ - debug: msg="this is a sub-block in a rescue"
+ always:
+ - debug: msg="this is an always task"
+ - block:
+ - debug: msg="this is a sub-block in an always"
+ post_tasks:
+ - debug: msg="this is a post_task"
+ """,
+ '/etc/ansible/roles/test_role/tasks/main.yml': """
+ - name: role task
+ debug: msg="this is a role task"
+ - block:
+ - name: role block task
+ debug: msg="inside block in role"
+ always:
+ - name: role always task
+ debug: msg="always task in block in role"
+ - include: foo.yml
+ - name: role task after include
+ debug: msg="after include in role"
+ - block:
+ - name: starting role nested block 1
+ debug:
+ - block:
+ - name: role nested block 1 task 1
+ debug:
+ - name: role nested block 1 task 2
+ debug:
+ - name: role nested block 1 task 3
+ debug:
+ - name: end of role nested block 1
+ debug:
+ - name: starting role nested block 2
+ debug:
+ - block:
+ - name: role nested block 2 task 1
+ debug:
+ - name: role nested block 2 task 2
+ debug:
+ - name: role nested block 2 task 3
+ debug:
+ - name: end of role nested block 2
+ debug:
+ """,
+ '/etc/ansible/roles/test_role/tasks/foo.yml': """
+ - name: role included task
+ debug: msg="this is task in an include from a role"
+ """
+ })
+
+ mock_var_manager = MagicMock()
+ mock_var_manager._fact_cache = dict()
+ mock_var_manager.get_vars.return_value = dict()
+
+ p = Playbook.load('test_play.yml', loader=fake_loader, variable_manager=mock_var_manager)
+
+ hosts = []
+ for i in range(0, 10):
+ host = MagicMock()
+ host.name = host.get_name.return_value = 'host%02d' % i
+ hosts.append(host)
+
+ mock_var_manager._fact_cache['host00'] = dict()
+
+ inventory = MagicMock()
+ inventory.get_hosts.return_value = hosts
+ inventory.filter_hosts.return_value = hosts
+
+ play_context = PlayContext(play=p._entries[0])
+
+ itr = PlayIterator(
+ inventory=inventory,
+ play=p._entries[0],
+ play_context=play_context,
+ variable_manager=mock_var_manager,
+ all_vars=dict(),
+ )
+
+ # pre task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ # implicit meta: flush_handlers
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ # role task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.name, "role task")
+ self.assertIsNotNone(task._role)
+ # role block task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role block task")
+ self.assertIsNotNone(task._role)
+ # role block always task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role always task")
+ self.assertIsNotNone(task._role)
+ # role include task
+ # (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ # self.assertIsNotNone(task)
+ # self.assertEqual(task.action, 'debug')
+ # self.assertEqual(task.name, "role included task")
+ # self.assertIsNotNone(task._role)
+ # role task after include
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role task after include")
+ self.assertIsNotNone(task._role)
+ # role nested block tasks
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "starting role nested block 1")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role nested block 1 task 1")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role nested block 1 task 2")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role nested block 1 task 3")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "end of role nested block 1")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "starting role nested block 2")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role nested block 2 task 1")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role nested block 2 task 2")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "role nested block 2 task 3")
+ self.assertIsNotNone(task._role)
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.name, "end of role nested block 2")
+ self.assertIsNotNone(task._role)
+ # implicit meta: role_complete
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ self.assertIsNotNone(task._role)
+ # regular play task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertIsNone(task._role)
+ # block task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg="this is a block task"))
+ # sub-block task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg="this is a sub-block in a block"))
+ # mark the host failed
+ itr.mark_host_failed(hosts[0])
+ # block rescue task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg="this is a rescue task"))
+ # sub-block rescue task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg="this is a sub-block in a rescue"))
+ # block always task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg="this is an always task"))
+ # sub-block always task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg="this is a sub-block in an always"))
+ # implicit meta: flush_handlers
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ # post task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ # implicit meta: flush_handlers
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ # end of iteration
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNone(task)
+
+ # host 0 shouldn't be in the failed hosts, as the error
+ # was handled by a rescue block
+ failed_hosts = itr.get_failed_hosts()
+ self.assertNotIn(hosts[0], failed_hosts)
+
+ def test_play_iterator_nested_blocks(self):
+ fake_loader = DictDataLoader({
+ "test_play.yml": """
+ - hosts: all
+ gather_facts: false
+ tasks:
+ - block:
+ - block:
+ - block:
+ - block:
+ - block:
+ - debug: msg="this is the first task"
+ - ping:
+ rescue:
+ - block:
+ - block:
+ - block:
+ - block:
+ - debug: msg="this is the rescue task"
+ always:
+ - block:
+ - block:
+ - block:
+ - block:
+ - debug: msg="this is the always task"
+ """,
+ })
+
+ mock_var_manager = MagicMock()
+ mock_var_manager._fact_cache = dict()
+ mock_var_manager.get_vars.return_value = dict()
+
+ p = Playbook.load('test_play.yml', loader=fake_loader, variable_manager=mock_var_manager)
+
+ hosts = []
+ for i in range(0, 10):
+ host = MagicMock()
+ host.name = host.get_name.return_value = 'host%02d' % i
+ hosts.append(host)
+
+ inventory = MagicMock()
+ inventory.get_hosts.return_value = hosts
+ inventory.filter_hosts.return_value = hosts
+
+ play_context = PlayContext(play=p._entries[0])
+
+ itr = PlayIterator(
+ inventory=inventory,
+ play=p._entries[0],
+ play_context=play_context,
+ variable_manager=mock_var_manager,
+ all_vars=dict(),
+ )
+
+ # implicit meta: flush_handlers
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ self.assertEqual(task.args, dict(_raw_params='flush_handlers'))
+ # get the first task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg='this is the first task'))
+ # fail the host
+ itr.mark_host_failed(hosts[0])
+ # get the resuce task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg='this is the rescue task'))
+ # get the always task
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'debug')
+ self.assertEqual(task.args, dict(msg='this is the always task'))
+ # implicit meta: flush_handlers
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ self.assertEqual(task.args, dict(_raw_params='flush_handlers'))
+ # implicit meta: flush_handlers
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNotNone(task)
+ self.assertEqual(task.action, 'meta')
+ self.assertEqual(task.args, dict(_raw_params='flush_handlers'))
+ # end of iteration
+ (host_state, task) = itr.get_next_task_for_host(hosts[0])
+ self.assertIsNone(task)
+
+ def test_play_iterator_add_tasks(self):
+ fake_loader = DictDataLoader({
+ 'test_play.yml': """
+ - hosts: all
+ gather_facts: no
+ tasks:
+ - debug: msg="dummy task"
+ """,
+ })
+
+ mock_var_manager = MagicMock()
+ mock_var_manager._fact_cache = dict()
+ mock_var_manager.get_vars.return_value = dict()
+
+ p = Playbook.load('test_play.yml', loader=fake_loader, variable_manager=mock_var_manager)
+
+ hosts = []
+ for i in range(0, 10):
+ host = MagicMock()
+ host.name = host.get_name.return_value = 'host%02d' % i
+ hosts.append(host)
+
+ inventory = MagicMock()
+ inventory.get_hosts.return_value = hosts
+ inventory.filter_hosts.return_value = hosts
+
+ play_context = PlayContext(play=p._entries[0])
+
+ itr = PlayIterator(
+ inventory=inventory,
+ play=p._entries[0],
+ play_context=play_context,
+ variable_manager=mock_var_manager,
+ all_vars=dict(),
+ )
+
+ # test the high-level add_tasks() method
+ s = HostState(blocks=[0, 1, 2])
+ itr._insert_tasks_into_state = MagicMock(return_value=s)
+ itr.add_tasks(hosts[0], [MagicMock(), MagicMock(), MagicMock()])
+ self.assertEqual(itr._host_states[hosts[0].name], s)
+
+ # now actually test the lower-level method that does the work
+ itr = PlayIterator(
+ inventory=inventory,
+ play=p._entries[0],
+ play_context=play_context,
+ variable_manager=mock_var_manager,
+ all_vars=dict(),
+ )
+
+ # iterate past first task
+ _, task = itr.get_next_task_for_host(hosts[0])
+ while (task and task.action != 'debug'):
+ _, task = itr.get_next_task_for_host(hosts[0])
+
+ if task is None:
+ raise Exception("iterated past end of play while looking for place to insert tasks")
+
+ # get the current host state and copy it so we can mutate it
+ s = itr.get_host_state(hosts[0])
+ s_copy = s.copy()
+
+ # assert with an empty task list, or if we're in a failed state, we simply return the state as-is
+ res_state = itr._insert_tasks_into_state(s_copy, task_list=[])
+ self.assertEqual(res_state, s_copy)
+
+ s_copy.fail_state = FailedStates.TASKS
+ res_state = itr._insert_tasks_into_state(s_copy, task_list=[MagicMock()])
+ self.assertEqual(res_state, s_copy)
+
+ # but if we've failed with a rescue/always block
+ mock_task = MagicMock()
+ s_copy.run_state = IteratingStates.RESCUE
+ res_state = itr._insert_tasks_into_state(s_copy, task_list=[mock_task])
+ self.assertEqual(res_state, s_copy)
+ self.assertIn(mock_task, res_state._blocks[res_state.cur_block].rescue)
+ itr.set_state_for_host(hosts[0].name, res_state)
+ (next_state, next_task) = itr.get_next_task_for_host(hosts[0], peek=True)
+ self.assertEqual(next_task, mock_task)
+ itr.set_state_for_host(hosts[0].name, s)
+
+ # test a regular insertion
+ s_copy = s.copy()
+ res_state = itr._insert_tasks_into_state(s_copy, task_list=[MagicMock()])
diff --git a/test/units/executor/test_playbook_executor.py b/test/units/executor/test_playbook_executor.py
new file mode 100644
index 0000000..6032dbb
--- /dev/null
+++ b/test/units/executor/test_playbook_executor.py
@@ -0,0 +1,148 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from units.compat import unittest
+from unittest.mock import MagicMock
+
+from ansible.executor.playbook_executor import PlaybookExecutor
+from ansible.playbook import Playbook
+from ansible.template import Templar
+from ansible.utils import context_objects as co
+
+from units.mock.loader import DictDataLoader
+
+
+class TestPlaybookExecutor(unittest.TestCase):
+
+ def setUp(self):
+ # Reset command line args for every test
+ co.GlobalCLIArgs._Singleton__instance = None
+
+ def tearDown(self):
+ # And cleanup after ourselves too
+ co.GlobalCLIArgs._Singleton__instance = None
+
+ def test_get_serialized_batches(self):
+ fake_loader = DictDataLoader({
+ 'no_serial.yml': '''
+ - hosts: all
+ gather_facts: no
+ tasks:
+ - debug: var=inventory_hostname
+ ''',
+ 'serial_int.yml': '''
+ - hosts: all
+ gather_facts: no
+ serial: 2
+ tasks:
+ - debug: var=inventory_hostname
+ ''',
+ 'serial_pct.yml': '''
+ - hosts: all
+ gather_facts: no
+ serial: 20%
+ tasks:
+ - debug: var=inventory_hostname
+ ''',
+ 'serial_list.yml': '''
+ - hosts: all
+ gather_facts: no
+ serial: [1, 2, 3]
+ tasks:
+ - debug: var=inventory_hostname
+ ''',
+ 'serial_list_mixed.yml': '''
+ - hosts: all
+ gather_facts: no
+ serial: [1, "20%", -1]
+ tasks:
+ - debug: var=inventory_hostname
+ ''',
+ })
+
+ mock_inventory = MagicMock()
+ mock_var_manager = MagicMock()
+
+ templar = Templar(loader=fake_loader)
+
+ pbe = PlaybookExecutor(
+ playbooks=['no_serial.yml', 'serial_int.yml', 'serial_pct.yml', 'serial_list.yml', 'serial_list_mixed.yml'],
+ inventory=mock_inventory,
+ variable_manager=mock_var_manager,
+ loader=fake_loader,
+ passwords=[],
+ )
+
+ playbook = Playbook.load(pbe._playbooks[0], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']
+ self.assertEqual(pbe._get_serialized_batches(play), [['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']])
+
+ playbook = Playbook.load(pbe._playbooks[1], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']
+ self.assertEqual(
+ pbe._get_serialized_batches(play),
+ [['host0', 'host1'], ['host2', 'host3'], ['host4', 'host5'], ['host6', 'host7'], ['host8', 'host9']]
+ )
+
+ playbook = Playbook.load(pbe._playbooks[2], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']
+ self.assertEqual(
+ pbe._get_serialized_batches(play),
+ [['host0', 'host1'], ['host2', 'host3'], ['host4', 'host5'], ['host6', 'host7'], ['host8', 'host9']]
+ )
+
+ playbook = Playbook.load(pbe._playbooks[3], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']
+ self.assertEqual(
+ pbe._get_serialized_batches(play),
+ [['host0'], ['host1', 'host2'], ['host3', 'host4', 'host5'], ['host6', 'host7', 'host8'], ['host9']]
+ )
+
+ playbook = Playbook.load(pbe._playbooks[4], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']
+ self.assertEqual(pbe._get_serialized_batches(play), [['host0'], ['host1', 'host2'], ['host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9']])
+
+ # Test when serial percent is under 1.0
+ playbook = Playbook.load(pbe._playbooks[2], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2']
+ self.assertEqual(pbe._get_serialized_batches(play), [['host0'], ['host1'], ['host2']])
+
+ # Test when there is a remainder for serial as a percent
+ playbook = Playbook.load(pbe._playbooks[2], variable_manager=mock_var_manager, loader=fake_loader)
+ play = playbook.get_plays()[0]
+ play.post_validate(templar)
+ mock_inventory.get_hosts.return_value = ['host0', 'host1', 'host2', 'host3', 'host4', 'host5', 'host6', 'host7', 'host8', 'host9', 'host10']
+ self.assertEqual(
+ pbe._get_serialized_batches(play),
+ [['host0', 'host1'], ['host2', 'host3'], ['host4', 'host5'], ['host6', 'host7'], ['host8', 'host9'], ['host10']]
+ )
diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py
new file mode 100644
index 0000000..315d26a
--- /dev/null
+++ b/test/units/executor/test_task_executor.py
@@ -0,0 +1,489 @@
+# (c) 2012-2014, Michael DeHaan <michael.dehaan@gmail.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from unittest import mock
+
+from units.compat import unittest
+from unittest.mock import patch, MagicMock
+from ansible.errors import AnsibleError
+from ansible.executor.task_executor import TaskExecutor, remove_omit
+from ansible.plugins.loader import action_loader, lookup_loader, module_loader
+from ansible.parsing.yaml.objects import AnsibleUnicode
+from ansible.utils.unsafe_proxy import AnsibleUnsafeText, AnsibleUnsafeBytes
+from ansible.module_utils.six import text_type
+
+from collections import namedtuple
+from units.mock.loader import DictDataLoader
+
+
+get_with_context_result = namedtuple('get_with_context_result', ['object', 'plugin_load_context'])
+
+
+class TestTaskExecutor(unittest.TestCase):
+
+ def test_task_executor_init(self):
+ fake_loader = DictDataLoader({})
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+ mock_play_context = MagicMock()
+ mock_shared_loader = MagicMock()
+ new_stdin = None
+ job_vars = dict()
+ mock_queue = MagicMock()
+ te = TaskExecutor(
+ host=mock_host,
+ task=mock_task,
+ job_vars=job_vars,
+ play_context=mock_play_context,
+ new_stdin=new_stdin,
+ loader=fake_loader,
+ shared_loader_obj=mock_shared_loader,
+ final_q=mock_queue,
+ )
+
+ def test_task_executor_run(self):
+ fake_loader = DictDataLoader({})
+
+ mock_host = MagicMock()
+
+ mock_task = MagicMock()
+ mock_task._role._role_path = '/path/to/role/foo'
+
+ mock_play_context = MagicMock()
+
+ mock_shared_loader = MagicMock()
+ mock_queue = MagicMock()
+
+ new_stdin = None
+ job_vars = dict()
+
+ te = TaskExecutor(
+ host=mock_host,
+ task=mock_task,
+ job_vars=job_vars,
+ play_context=mock_play_context,
+ new_stdin=new_stdin,
+ loader=fake_loader,
+ shared_loader_obj=mock_shared_loader,
+ final_q=mock_queue,
+ )
+
+ te._get_loop_items = MagicMock(return_value=None)
+ te._execute = MagicMock(return_value=dict())
+ res = te.run()
+
+ te._get_loop_items = MagicMock(return_value=[])
+ res = te.run()
+
+ te._get_loop_items = MagicMock(return_value=['a', 'b', 'c'])
+ te._run_loop = MagicMock(return_value=[dict(item='a', changed=True), dict(item='b', failed=True), dict(item='c')])
+ res = te.run()
+
+ te._get_loop_items = MagicMock(side_effect=AnsibleError(""))
+ res = te.run()
+ self.assertIn("failed", res)
+
+ def test_task_executor_run_clean_res(self):
+ te = TaskExecutor(None, MagicMock(), None, None, None, None, None, None)
+ te._get_loop_items = MagicMock(return_value=[1])
+ te._run_loop = MagicMock(
+ return_value=[
+ {
+ 'unsafe_bytes': AnsibleUnsafeBytes(b'{{ $bar }}'),
+ 'unsafe_text': AnsibleUnsafeText(u'{{ $bar }}'),
+ 'bytes': b'bytes',
+ 'text': u'text',
+ 'int': 1,
+ }
+ ]
+ )
+ res = te.run()
+ data = res['results'][0]
+ self.assertIsInstance(data['unsafe_bytes'], AnsibleUnsafeText)
+ self.assertIsInstance(data['unsafe_text'], AnsibleUnsafeText)
+ self.assertIsInstance(data['bytes'], text_type)
+ self.assertIsInstance(data['text'], text_type)
+ self.assertIsInstance(data['int'], int)
+
+ def test_task_executor_get_loop_items(self):
+ fake_loader = DictDataLoader({})
+
+ mock_host = MagicMock()
+
+ mock_task = MagicMock()
+ mock_task.loop_with = 'items'
+ mock_task.loop = ['a', 'b', 'c']
+
+ mock_play_context = MagicMock()
+
+ mock_shared_loader = MagicMock()
+ mock_shared_loader.lookup_loader = lookup_loader
+
+ new_stdin = None
+ job_vars = dict()
+ mock_queue = MagicMock()
+
+ te = TaskExecutor(
+ host=mock_host,
+ task=mock_task,
+ job_vars=job_vars,
+ play_context=mock_play_context,
+ new_stdin=new_stdin,
+ loader=fake_loader,
+ shared_loader_obj=mock_shared_loader,
+ final_q=mock_queue,
+ )
+
+ items = te._get_loop_items()
+ self.assertEqual(items, ['a', 'b', 'c'])
+
+ def test_task_executor_run_loop(self):
+ items = ['a', 'b', 'c']
+
+ fake_loader = DictDataLoader({})
+
+ mock_host = MagicMock()
+
+ def _copy(exclude_parent=False, exclude_tasks=False):
+ new_item = MagicMock()
+ return new_item
+
+ mock_task = MagicMock()
+ mock_task.copy.side_effect = _copy
+
+ mock_play_context = MagicMock()
+
+ mock_shared_loader = MagicMock()
+ mock_queue = MagicMock()
+
+ new_stdin = None
+ job_vars = dict()
+
+ te = TaskExecutor(
+ host=mock_host,
+ task=mock_task,
+ job_vars=job_vars,
+ play_context=mock_play_context,
+ new_stdin=new_stdin,
+ loader=fake_loader,
+ shared_loader_obj=mock_shared_loader,
+ final_q=mock_queue,
+ )
+
+ def _execute(variables):
+ return dict(item=variables.get('item'))
+
+ te._execute = MagicMock(side_effect=_execute)
+
+ res = te._run_loop(items)
+ self.assertEqual(len(res), 3)
+
+ def test_task_executor_get_action_handler(self):
+ te = TaskExecutor(
+ host=MagicMock(),
+ task=MagicMock(),
+ job_vars={},
+ play_context=MagicMock(),
+ new_stdin=None,
+ loader=DictDataLoader({}),
+ shared_loader_obj=MagicMock(),
+ final_q=MagicMock(),
+ )
+
+ context = MagicMock(resolved=False)
+ te._shared_loader_obj.module_loader.find_plugin_with_context.return_value = context
+ action_loader = te._shared_loader_obj.action_loader
+ action_loader.has_plugin.return_value = True
+ action_loader.get.return_value = mock.sentinel.handler
+
+ mock_connection = MagicMock()
+ mock_templar = MagicMock()
+ action = 'namespace.prefix_suffix'
+ te._task.action = action
+
+ handler = te._get_action_handler(mock_connection, mock_templar)
+
+ self.assertIs(mock.sentinel.handler, handler)
+
+ action_loader.has_plugin.assert_called_once_with(
+ action, collection_list=te._task.collections)
+
+ action_loader.get.assert_called_once_with(
+ te._task.action, task=te._task, connection=mock_connection,
+ play_context=te._play_context, loader=te._loader,
+ templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
+ collection_list=te._task.collections)
+
+ def test_task_executor_get_handler_prefix(self):
+ te = TaskExecutor(
+ host=MagicMock(),
+ task=MagicMock(),
+ job_vars={},
+ play_context=MagicMock(),
+ new_stdin=None,
+ loader=DictDataLoader({}),
+ shared_loader_obj=MagicMock(),
+ final_q=MagicMock(),
+ )
+
+ context = MagicMock(resolved=False)
+ te._shared_loader_obj.module_loader.find_plugin_with_context.return_value = context
+ action_loader = te._shared_loader_obj.action_loader
+ action_loader.has_plugin.side_effect = [False, True]
+ action_loader.get.return_value = mock.sentinel.handler
+ action_loader.__contains__.return_value = True
+
+ mock_connection = MagicMock()
+ mock_templar = MagicMock()
+ action = 'namespace.netconf_suffix'
+ module_prefix = action.split('_', 1)[0]
+ te._task.action = action
+
+ handler = te._get_action_handler(mock_connection, mock_templar)
+
+ self.assertIs(mock.sentinel.handler, handler)
+ action_loader.has_plugin.assert_has_calls([mock.call(action, collection_list=te._task.collections), # called twice
+ mock.call(module_prefix, collection_list=te._task.collections)])
+
+ action_loader.get.assert_called_once_with(
+ module_prefix, task=te._task, connection=mock_connection,
+ play_context=te._play_context, loader=te._loader,
+ templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
+ collection_list=te._task.collections)
+
+ def test_task_executor_get_handler_normal(self):
+ te = TaskExecutor(
+ host=MagicMock(),
+ task=MagicMock(),
+ job_vars={},
+ play_context=MagicMock(),
+ new_stdin=None,
+ loader=DictDataLoader({}),
+ shared_loader_obj=MagicMock(),
+ final_q=MagicMock(),
+ )
+
+ action_loader = te._shared_loader_obj.action_loader
+ action_loader.has_plugin.return_value = False
+ action_loader.get.return_value = mock.sentinel.handler
+ action_loader.__contains__.return_value = False
+ module_loader = te._shared_loader_obj.module_loader
+ context = MagicMock(resolved=False)
+ module_loader.find_plugin_with_context.return_value = context
+
+ mock_connection = MagicMock()
+ mock_templar = MagicMock()
+ action = 'namespace.prefix_suffix'
+ module_prefix = action.split('_', 1)[0]
+ te._task.action = action
+ handler = te._get_action_handler(mock_connection, mock_templar)
+
+ self.assertIs(mock.sentinel.handler, handler)
+
+ action_loader.has_plugin.assert_has_calls([mock.call(action, collection_list=te._task.collections),
+ mock.call(module_prefix, collection_list=te._task.collections)])
+
+ action_loader.get.assert_called_once_with(
+ 'ansible.legacy.normal', task=te._task, connection=mock_connection,
+ play_context=te._play_context, loader=te._loader,
+ templar=mock_templar, shared_loader_obj=te._shared_loader_obj,
+ collection_list=None)
+
+ def test_task_executor_execute(self):
+ fake_loader = DictDataLoader({})
+
+ mock_host = MagicMock()
+
+ mock_task = MagicMock()
+ mock_task.action = 'mock.action'
+ mock_task.args = dict()
+ mock_task.become = False
+ mock_task.retries = 0
+ mock_task.delay = -1
+ mock_task.register = 'foo'
+ mock_task.until = None
+ mock_task.changed_when = None
+ mock_task.failed_when = None
+ mock_task.post_validate.return_value = None
+ # mock_task.async_val cannot be left unset, because on Python 3 MagicMock()
+ # > 0 raises a TypeError There are two reasons for using the value 1
+ # here: on Python 2 comparing MagicMock() > 0 returns True, and the
+ # other reason is that if I specify 0 here, the test fails. ;)
+ mock_task.async_val = 1
+ mock_task.poll = 0
+
+ mock_play_context = MagicMock()
+ mock_play_context.post_validate.return_value = None
+ mock_play_context.update_vars.return_value = None
+
+ mock_connection = MagicMock()
+ mock_connection.force_persistence = False
+ mock_connection.supports_persistence = False
+ mock_connection.set_host_overrides.return_value = None
+ mock_connection._connect.return_value = None
+
+ mock_action = MagicMock()
+ mock_queue = MagicMock()
+
+ shared_loader = MagicMock()
+ new_stdin = None
+ job_vars = dict(omit="XXXXXXXXXXXXXXXXXXX")
+
+ te = TaskExecutor(
+ host=mock_host,
+ task=mock_task,
+ job_vars=job_vars,
+ play_context=mock_play_context,
+ new_stdin=new_stdin,
+ loader=fake_loader,
+ shared_loader_obj=shared_loader,
+ final_q=mock_queue,
+ )
+
+ te._get_connection = MagicMock(return_value=mock_connection)
+ context = MagicMock()
+ te._get_action_handler_with_context = MagicMock(return_value=get_with_context_result(mock_action, context))
+
+ mock_action.run.return_value = dict(ansible_facts=dict())
+ res = te._execute()
+
+ mock_task.changed_when = MagicMock(return_value=AnsibleUnicode("1 == 1"))
+ res = te._execute()
+
+ mock_task.changed_when = None
+ mock_task.failed_when = MagicMock(return_value=AnsibleUnicode("1 == 1"))
+ res = te._execute()
+
+ mock_task.failed_when = None
+ mock_task.evaluate_conditional.return_value = False
+ res = te._execute()
+
+ mock_task.evaluate_conditional.return_value = True
+ mock_task.args = dict(_raw_params='foo.yml', a='foo', b='bar')
+ mock_task.action = 'include'
+ res = te._execute()
+
+ def test_task_executor_poll_async_result(self):
+ fake_loader = DictDataLoader({})
+
+ mock_host = MagicMock()
+
+ mock_task = MagicMock()
+ mock_task.async_val = 0.1
+ mock_task.poll = 0.05
+
+ mock_play_context = MagicMock()
+
+ mock_connection = MagicMock()
+
+ mock_action = MagicMock()
+ mock_queue = MagicMock()
+
+ shared_loader = MagicMock()
+ shared_loader.action_loader = action_loader
+
+ new_stdin = None
+ job_vars = dict(omit="XXXXXXXXXXXXXXXXXXX")
+
+ te = TaskExecutor(
+ host=mock_host,
+ task=mock_task,
+ job_vars=job_vars,
+ play_context=mock_play_context,
+ new_stdin=new_stdin,
+ loader=fake_loader,
+ shared_loader_obj=shared_loader,
+ final_q=mock_queue,
+ )
+
+ te._connection = MagicMock()
+
+ def _get(*args, **kwargs):
+ mock_action = MagicMock()
+ mock_action.run.return_value = dict(stdout='')
+ return mock_action
+
+ # testing with some bad values in the result passed to poll async,
+ # and with a bad value returned from the mock action
+ with patch.object(action_loader, 'get', _get):
+ mock_templar = MagicMock()
+ res = te._poll_async_result(result=dict(), templar=mock_templar)
+ self.assertIn('failed', res)
+ res = te._poll_async_result(result=dict(ansible_job_id=1), templar=mock_templar)
+ self.assertIn('failed', res)
+
+ def _get(*args, **kwargs):
+ mock_action = MagicMock()
+ mock_action.run.return_value = dict(finished=1)
+ return mock_action
+
+ # now testing with good values
+ with patch.object(action_loader, 'get', _get):
+ mock_templar = MagicMock()
+ res = te._poll_async_result(result=dict(ansible_job_id=1), templar=mock_templar)
+ self.assertEqual(res, dict(finished=1))
+
+ def test_recursive_remove_omit(self):
+ omit_token = 'POPCORN'
+
+ data = {
+ 'foo': 'bar',
+ 'baz': 1,
+ 'qux': ['one', 'two', 'three'],
+ 'subdict': {
+ 'remove': 'POPCORN',
+ 'keep': 'not_popcorn',
+ 'subsubdict': {
+ 'remove': 'POPCORN',
+ 'keep': 'not_popcorn',
+ },
+ 'a_list': ['POPCORN'],
+ },
+ 'a_list': ['POPCORN'],
+ 'list_of_lists': [
+ ['some', 'thing'],
+ ],
+ 'list_of_dicts': [
+ {
+ 'remove': 'POPCORN',
+ }
+ ],
+ }
+
+ expected = {
+ 'foo': 'bar',
+ 'baz': 1,
+ 'qux': ['one', 'two', 'three'],
+ 'subdict': {
+ 'keep': 'not_popcorn',
+ 'subsubdict': {
+ 'keep': 'not_popcorn',
+ },
+ 'a_list': ['POPCORN'],
+ },
+ 'a_list': ['POPCORN'],
+ 'list_of_lists': [
+ ['some', 'thing'],
+ ],
+ 'list_of_dicts': [{}],
+ }
+
+ self.assertEqual(remove_omit(data, omit_token), expected)
diff --git a/test/units/executor/test_task_queue_manager_callbacks.py b/test/units/executor/test_task_queue_manager_callbacks.py
new file mode 100644
index 0000000..c63385d
--- /dev/null
+++ b/test/units/executor/test_task_queue_manager_callbacks.py
@@ -0,0 +1,121 @@
+# (c) 2016, Steve Kuznetsov <skuznets@redhat.com>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+
+from units.compat import unittest
+from unittest.mock import MagicMock
+
+from ansible.executor.task_queue_manager import TaskQueueManager
+from ansible.playbook import Playbook
+from ansible.plugins.callback import CallbackBase
+from ansible.utils import context_objects as co
+
+__metaclass__ = type
+
+
+class TestTaskQueueManagerCallbacks(unittest.TestCase):
+ def setUp(self):
+ inventory = MagicMock()
+ variable_manager = MagicMock()
+ loader = MagicMock()
+ passwords = []
+
+ # Reset the stored command line args
+ co.GlobalCLIArgs._Singleton__instance = None
+ self._tqm = TaskQueueManager(inventory, variable_manager, loader, passwords)
+ self._playbook = Playbook(loader)
+
+ # we use a MagicMock to register the result of the call we
+ # expect to `v2_playbook_on_call`. We don't mock out the
+ # method since we're testing code that uses `inspect` to
+ # look at that method's argspec and we want to ensure this
+ # test is easy to reason about.
+ self._register = MagicMock()
+
+ def tearDown(self):
+ # Reset the stored command line args
+ co.GlobalCLIArgs._Singleton__instance = None
+
+ def test_task_queue_manager_callbacks_v2_playbook_on_start(self):
+ """
+ Assert that no exceptions are raised when sending a Playbook
+ start callback to a current callback module plugin.
+ """
+ register = self._register
+
+ class CallbackModule(CallbackBase):
+ """
+ This is a callback module with the current
+ method signature for `v2_playbook_on_start`.
+ """
+ CALLBACK_VERSION = 2.0
+ CALLBACK_TYPE = 'notification'
+ CALLBACK_NAME = 'current_module'
+
+ def v2_playbook_on_start(self, playbook):
+ register(self, playbook)
+
+ callback_module = CallbackModule()
+ self._tqm._callback_plugins.append(callback_module)
+ self._tqm.send_callback('v2_playbook_on_start', self._playbook)
+ register.assert_called_once_with(callback_module, self._playbook)
+
+ def test_task_queue_manager_callbacks_v2_playbook_on_start_wrapped(self):
+ """
+ Assert that no exceptions are raised when sending a Playbook
+ start callback to a wrapped current callback module plugin.
+ """
+ register = self._register
+
+ def wrap_callback(func):
+ """
+ This wrapper changes the exposed argument
+ names for a method from the original names
+ to (*args, **kwargs). This is used in order
+ to validate that wrappers which change par-
+ ameter names do not break the TQM callback
+ system.
+
+ :param func: function to decorate
+ :return: decorated function
+ """
+
+ def wrapper(*args, **kwargs):
+ return func(*args, **kwargs)
+
+ return wrapper
+
+ class WrappedCallbackModule(CallbackBase):
+ """
+ This is a callback module with the current
+ method signature for `v2_playbook_on_start`
+ wrapped in order to change the signature.
+ """
+ CALLBACK_VERSION = 2.0
+ CALLBACK_TYPE = 'notification'
+ CALLBACK_NAME = 'current_module'
+
+ @wrap_callback
+ def v2_playbook_on_start(self, playbook):
+ register(self, playbook)
+
+ callback_module = WrappedCallbackModule()
+ self._tqm._callback_plugins.append(callback_module)
+ self._tqm.send_callback('v2_playbook_on_start', self._playbook)
+ register.assert_called_once_with(callback_module, self._playbook)
diff --git a/test/units/executor/test_task_result.py b/test/units/executor/test_task_result.py
new file mode 100644
index 0000000..8b79571
--- /dev/null
+++ b/test/units/executor/test_task_result.py
@@ -0,0 +1,171 @@
+# (c) 2016, James Cammarata <jimi@sngx.net>
+#
+# This file is part of Ansible
+#
+# Ansible 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.
+#
+# Ansible 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 Ansible. If not, see <http://www.gnu.org/licenses/>.
+
+# Make coding more python3-ish
+from __future__ import (absolute_import, division, print_function)
+__metaclass__ = type
+
+from units.compat import unittest
+from unittest.mock import patch, MagicMock
+
+from ansible.executor.task_result import TaskResult
+
+
+class TestTaskResult(unittest.TestCase):
+ def test_task_result_basic(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # test loading a result with a dict
+ tr = TaskResult(mock_host, mock_task, dict())
+
+ # test loading a result with a JSON string
+ with patch('ansible.parsing.dataloader.DataLoader.load') as p:
+ tr = TaskResult(mock_host, mock_task, '{}')
+
+ def test_task_result_is_changed(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # test with no changed in result
+ tr = TaskResult(mock_host, mock_task, dict())
+ self.assertFalse(tr.is_changed())
+
+ # test with changed in the result
+ tr = TaskResult(mock_host, mock_task, dict(changed=True))
+ self.assertTrue(tr.is_changed())
+
+ # test with multiple results but none changed
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]))
+ self.assertFalse(tr.is_changed())
+
+ # test with multiple results and one changed
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(changed=False), dict(changed=True), dict(some_key=False)]))
+ self.assertTrue(tr.is_changed())
+
+ def test_task_result_is_skipped(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # test with no skipped in result
+ tr = TaskResult(mock_host, mock_task, dict())
+ self.assertFalse(tr.is_skipped())
+
+ # test with skipped in the result
+ tr = TaskResult(mock_host, mock_task, dict(skipped=True))
+ self.assertTrue(tr.is_skipped())
+
+ # test with multiple results but none skipped
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]))
+ self.assertFalse(tr.is_skipped())
+
+ # test with multiple results and one skipped
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(skipped=False), dict(skipped=True), dict(some_key=False)]))
+ self.assertFalse(tr.is_skipped())
+
+ # test with multiple results and all skipped
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(skipped=True), dict(skipped=True), dict(skipped=True)]))
+ self.assertTrue(tr.is_skipped())
+
+ # test with multiple squashed results (list of strings)
+ # first with the main result having skipped=False
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=["a", "b", "c"], skipped=False))
+ self.assertFalse(tr.is_skipped())
+ # then with the main result having skipped=True
+ tr = TaskResult(mock_host, mock_task, dict(results=["a", "b", "c"], skipped=True))
+ self.assertTrue(tr.is_skipped())
+
+ def test_task_result_is_unreachable(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # test with no unreachable in result
+ tr = TaskResult(mock_host, mock_task, dict())
+ self.assertFalse(tr.is_unreachable())
+
+ # test with unreachable in the result
+ tr = TaskResult(mock_host, mock_task, dict(unreachable=True))
+ self.assertTrue(tr.is_unreachable())
+
+ # test with multiple results but none unreachable
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(foo='bar'), dict(bam='baz'), True]))
+ self.assertFalse(tr.is_unreachable())
+
+ # test with multiple results and one unreachable
+ mock_task.loop = 'foo'
+ tr = TaskResult(mock_host, mock_task, dict(results=[dict(unreachable=False), dict(unreachable=True), dict(some_key=False)]))
+ self.assertTrue(tr.is_unreachable())
+
+ def test_task_result_is_failed(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # test with no failed in result
+ tr = TaskResult(mock_host, mock_task, dict())
+ self.assertFalse(tr.is_failed())
+
+ # test failed result with rc values (should not matter)
+ tr = TaskResult(mock_host, mock_task, dict(rc=0))
+ self.assertFalse(tr.is_failed())
+ tr = TaskResult(mock_host, mock_task, dict(rc=1))
+ self.assertFalse(tr.is_failed())
+
+ # test with failed in result
+ tr = TaskResult(mock_host, mock_task, dict(failed=True))
+ self.assertTrue(tr.is_failed())
+
+ # test with failed_when in result
+ tr = TaskResult(mock_host, mock_task, dict(failed_when_result=True))
+ self.assertTrue(tr.is_failed())
+
+ def test_task_result_no_log(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # no_log should remove secrets
+ tr = TaskResult(mock_host, mock_task, dict(_ansible_no_log=True, secret='DONTSHOWME'))
+ clean = tr.clean_copy()
+ self.assertTrue('secret' not in clean._result)
+
+ def test_task_result_no_log_preserve(self):
+ mock_host = MagicMock()
+ mock_task = MagicMock()
+
+ # no_log should not remove presrved keys
+ tr = TaskResult(
+ mock_host,
+ mock_task,
+ dict(
+ _ansible_no_log=True,
+ retries=5,
+ attempts=5,
+ changed=False,
+ foo='bar',
+ )
+ )
+ clean = tr.clean_copy()
+ self.assertTrue('retries' in clean._result)
+ self.assertTrue('attempts' in clean._result)
+ self.assertTrue('changed' in clean._result)
+ self.assertTrue('foo' not in clean._result)
generated by cgit v1.2.3 (git 2.39.1) at 2024-06-20 08:57:19 +0000