Adding unit tests for TaskExecutor (v2)

test/units/executor/test_task_executor.py

+# (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 ansible.compat.tests import unittest
+from ansible.compat.tests.mock import patch, MagicMock
+
+from ansible.errors import AnsibleError, AnsibleParserError
+from ansible.executor.connection_info import ConnectionInformation
+from ansible.executor.task_executor import TaskExecutor
+from ansible.plugins import action_loader
+
+from units.mock.loader import DictDataLoader
+
+class TestTaskExecutor(unittest.TestCase):
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    def test_task_executor_init(self):
+        fake_loader = DictDataLoader({})
+        mock_host = MagicMock()
+        mock_task = MagicMock()
+        mock_conn_info = MagicMock()
+        mock_shared_loader = MagicMock()
+        new_stdin = None
+        job_vars = dict()
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = mock_shared_loader,
+        )
+
+    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_conn_info = MagicMock()
+
+        mock_shared_loader = MagicMock()
+
+        new_stdin = None
+        job_vars = dict()
+
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = mock_shared_loader,
+        )
+
+        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_get_loop_items(self):
+        fake_loader = DictDataLoader({})
+        
+        mock_host = MagicMock()
+
+        mock_task = MagicMock()
+        mock_task.loop = 'items'
+        mock_task.loop_args = ['a', 'b', 'c']
+
+        mock_conn_info = MagicMock()
+
+        mock_shared_loader = MagicMock()
+
+        new_stdin = None
+        job_vars = dict()
+
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = mock_shared_loader,
+        )
+
+        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():
+            new_item = MagicMock()
+            return new_item
+
+        mock_task = MagicMock()
+        mock_task.copy.side_effect = _copy
+
+        mock_conn_info = MagicMock()
+
+        mock_shared_loader = MagicMock()
+
+        new_stdin = None
+        job_vars = dict()
+
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = mock_shared_loader,
+        )
+
+        def _execute(variables):
+            return dict(item=variables.get('item'))
+
+        te._squash_items = MagicMock(return_value=items)
+        te._execute = MagicMock(side_effect=_execute)
+
+        res = te._run_loop(items)
+        self.assertEqual(len(res), 3)
+
+    def test_task_executor_squash_items(self):
+        items = ['a', 'b', 'c']
+
+        fake_loader = DictDataLoader({})
+
+        mock_host = MagicMock()
+
+        def _evaluate_conditional(templar, variables):
+            item = variables.get('item')
+            if item == 'b':
+                return False
+            return True
+
+        mock_task = MagicMock()
+        mock_task.evaluate_conditional.side_effect = _evaluate_conditional
+
+        mock_conn_info = MagicMock()
+
+        mock_shared_loader = None
+
+        new_stdin = None
+        job_vars = dict()
+
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = mock_shared_loader,
+        )
+
+        mock_task.action = 'foo'
+        new_items = te._squash_items(items=items, variables=job_vars)
+        self.assertEqual(new_items, ['a', 'b', 'c'])
+
+        mock_task.action = 'yum'
+        new_items = te._squash_items(items=items, variables=job_vars)
+        self.assertEqual(new_items, ['a,c'])
+
+    def test_task_executor_execute(self):
+        fake_loader = DictDataLoader({})
+
+        mock_host = MagicMock()
+
+        mock_task = MagicMock()
+        mock_task.args = dict()
+        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_conn_info = MagicMock()
+        mock_conn_info.post_validate.return_value = None
+        mock_conn_info.update_vars.return_value = None
+
+        mock_connection = MagicMock()
+        mock_connection.set_host_overrides.return_value = None
+        mock_connection._connect.return_value = None
+
+        mock_action = MagicMock()
+
+        shared_loader = None
+        new_stdin = None
+        job_vars = dict(omit="XXXXXXXXXXXXXXXXXXX")
+
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = shared_loader,
+        )
+
+        te._get_connection = MagicMock(return_value=mock_connection)
+        te._get_action_handler = MagicMock(return_value=mock_action)
+
+        mock_action.run.return_value = dict(ansible_facts=dict())
+        res = te._execute()
+
+        mock_task.changed_when = "1 == 1"
+        res = te._execute()
+
+        mock_task.changed_when = None
+        mock_task.failed_when = "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 = 3
+        mock_task.poll  = 1
+
+        mock_conn_info = MagicMock()
+
+        mock_connection = MagicMock()
+
+        mock_action = MagicMock()
+
+        shared_loader = None
+        new_stdin = None
+        job_vars = dict(omit="XXXXXXXXXXXXXXXXXXX")
+
+        te = TaskExecutor(
+            host = mock_host,
+            task = mock_task,
+            job_vars = job_vars,
+            connection_info = mock_conn_info,
+            new_stdin = new_stdin,
+            loader = fake_loader,
+            shared_loader_obj = shared_loader,
+        )
+
+        te._connection = MagicMock()
+
+        def _get(*args, **kwargs):
+            mock_action = MagicMock()
+            mock_action.run.return_value = dict()
+            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))
+